StreamingContainerManager implements PlanContext { private void init(boolean enableEventRecording) { setupWsClient(); setupRecording(enableEventRecording); setupStringCodecs(); try { Path file = new Path(this.vars.appPath); URI uri = file.toUri(); Configuration config = new YarnConfiguration(); fileContext = uri.getScheme() == null ? FileContext.getFileContext(config) : FileContext.getFileContext(uri, config); saveMetaInfo(); String fileName = String.format(CONTAINERS_INFO_FILENAME_FORMAT, plan.getLogicalPlan().getValue(LogicalPlan.APPLICATION_ATTEMPT_ID)); this.containerFile = new FSJsonLineFile(fileContext, new Path(this.vars.appPath, fileName), FsPermission.getDefault()); this.containerFile.append(getAppMasterContainerInfo()); fileName = String.format(OPERATORS_INFO_FILENAME_FORMAT, plan.getLogicalPlan().getValue(LogicalPlan.APPLICATION_ATTEMPT_ID)); this.operatorFile = new FSJsonLineFile(fileContext, new Path(this.vars.appPath, fileName), FsPermission.getDefault()); } catch (IOException ex) { throw Throwables.propagate(ex); } } StreamingContainerManager(LogicalPlan dag, Clock clock);  StreamingContainerManager(LogicalPlan dag);  StreamingContainerManager(LogicalPlan dag, boolean enableEventRecording, Clock clock); private  StreamingContainerManager(CheckpointState checkpointedState, boolean enableEventRecording); Journal getJournal(); final ContainerInfo getAppMasterContainerInfo(); void updateRPCLatency(String containerId, long latency); void teardown(); void subscribeToEvents(Object listener); PhysicalPlan getPhysicalPlan(); long getCommittedWindowId(); boolean isGatewayConnected(); SharedPubSubWebSocketClient getWsClient(); List<AppDataSource> getAppDataSources(); Map<String, Map<String, Object>> getLatestLogicalMetrics(); void monitorHeartbeat(boolean waitForRecovery); Queue<Pair<Long, Map<String, Object>>> getWindowMetrics(String operatorName); int processEvents(); void scheduleContainerRestart(String containerId); void removeContainerAgent(String containerId); Collection<ContainerInfo> getCompletedContainerInfo(); StreamingContainerAgent assignContainer(ContainerResource resource, InetSocketAddress bufferServerAddr); StreamingContainerAgent getContainerAgent(String containerId); Collection<StreamingContainerAgent> getContainerAgents(); @SuppressWarnings("StatementWithEmptyBody") ContainerHeartbeatResponse processHeartbeat(ContainerHeartbeat heartbeat); long updateOperatorLatency(PTOperator oper, UpdateOperatorLatencyContext ctx); void updateRecoveryCheckpoints(PTOperator operator, UpdateCheckpointsContext ctx, boolean recovery); long windowIdToMillis(long windowId); long getWindowStartMillis(); void shutdownAllContainers(ShutdownType type, String message); void deployAfterRestart(); @Override void deploy(Set<PTContainer> releaseContainers, Collection<PTOperator> undeploy, Set<PTContainer> startContainers, Collection<PTOperator> deploy); @Override void recordEventAsync(StramEvent ev); @Override void dispatch(Runnable r); OperatorInfo getOperatorInfo(int operatorId); List<OperatorInfo> getOperatorInfoList(); LogicalOperatorInfo getLogicalOperatorInfo(String operatorName); ModuleMeta getModuleMeta(String moduleName); List<LogicalOperatorInfo> getLogicalOperatorInfoList(); OperatorAggregationInfo getOperatorAggregationInfo(String operatorName); static long toWsWindowId(long windowId); List<StreamInfo> getStreamInfoList(); void startRecording(String id, int operId, String portName, long numWindows); void stopRecording(int operId, String portName); void syncStats(); void syncEvents(); void stopContainer(String containerId); Recoverable getSetOperatorProperty(String operatorName, String propertyName, String propertyValue); Recoverable getSetPhysicalOperatorProperty(int operatorId, String propertyName, String propertyValue); void setOperatorProperty(String operatorName, String propertyName, String propertyValue); void setPhysicalOperatorProperty(int operatorId, String propertyName, String propertyValue); @Override void addOperatorRequest(PTOperator oper, StramToNodeRequest request); void setLoggersLevel(Map<String, String> changedLoggers); FutureTask<Object> getPhysicalOperatorProperty(int operatorId, String propertyName, long waitTime); Attribute.AttributeMap getApplicationAttributes(); Attribute.AttributeMap getOperatorAttributes(String operatorId); Attribute.AttributeMap getPortAttributes(String operatorId, String portName); LogicalPlan getLogicalPlan(); FutureTask<Object> logicalPlanModification(List<LogicalPlanRequest> requests); CriticalPathInfo getCriticalPathInfo(); @Override void writeJournal(Recoverable operation); static StreamingContainerManager getInstance(RecoveryHandler rh, LogicalPlan dag, boolean enableEventRecording); void setApexPluginDispatcher(ApexPluginDispatcher manager); static final String BUILTIN_APPDATA_URL; static final String CONTAINERS_INFO_FILENAME_FORMAT; static final String OPERATORS_INFO_FILENAME_FORMAT; static final String APP_META_FILENAME; static final String APP_META_KEY_ATTRIBUTES; static final String APP_META_KEY_METRICS; static final String EMBEDDABLE_QUERY_NAME_SUFFIX; static final Recoverable SET_OPERATOR_PROPERTY; static final Recoverable SET_PHYSICAL_OPERATOR_PROPERTY; static final int METRIC_QUEUE_SIZE; transient ApexPluginDispatcher apexPluginDispatcher; }
@Test public void testAppDataPush() throws Exception { if (StramTestSupport.isInTravis()) { LOG.info("Test testAppDataPush is disabled in Travis"); return; } final String topic = "xyz"; final List<String> messages = new ArrayList<>(); EmbeddedWebSocketServer server = new EmbeddedWebSocketServer(0); server.setWebSocket(new WebSocket.OnTextMessage() { @Override public void onMessage(String data) { messages.add(data); } @Override public void onOpen(WebSocket.Connection connection) { } @Override public void onClose(int closeCode, String message) { } }); try { server.start(); int port = server.getPort(); TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); dag.addStream("o1.outport", o1.outport, o2.inport1); dag.setAttribute(LogicalPlan.METRICS_TRANSPORT, new AutoMetricBuiltInTransport(topic)); dag.setAttribute(LogicalPlan.GATEWAY_CONNECT_ADDRESS, "localhost:" + port); dag.setAttribute(LogicalPlan.PUBSUB_CONNECT_TIMEOUT_MILLIS, 2000); StramLocalCluster lc = new StramLocalCluster(dag); StreamingContainerManager dnmgr = lc.dnmgr; StramAppContext appContext = new StramTestSupport.TestAppContext(dag.getAttributes()); AppDataPushAgent pushAgent = new AppDataPushAgent(dnmgr, appContext); pushAgent.init(); pushAgent.pushData(); Thread.sleep(1000); Assert.assertTrue(messages.size() > 0); pushAgent.close(); JSONObject message = new JSONObject(messages.get(0)); Assert.assertEquals(topic, message.getString("topic")); Assert.assertEquals("publish", message.getString("type")); JSONObject data = message.getJSONObject("data"); Assert.assertTrue(StringUtils.isNotBlank(data.getString("appId"))); Assert.assertTrue(StringUtils.isNotBlank(data.getString("appUser"))); Assert.assertTrue(StringUtils.isNotBlank(data.getString("appName"))); JSONObject logicalOperators = data.getJSONObject("logicalOperators"); for (String opName : new String[]{"o1", "o2"}) { JSONObject opObj = logicalOperators.getJSONObject(opName); Assert.assertTrue(opObj.has("totalTuplesProcessed")); Assert.assertTrue(opObj.has("totalTuplesEmitted")); Assert.assertTrue(opObj.has("tuplesProcessedPSMA")); Assert.assertTrue(opObj.has("tuplesEmittedPSMA")); Assert.assertTrue(opObj.has("latencyMA")); } } finally { server.stop(); } }
@Test public void testCustomMetricsTransport() throws Exception { TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); dag.addStream("o1.outport", o1.outport, o2.inport1); dag.setAttribute(LogicalPlan.METRICS_TRANSPORT, new TestMetricTransport("xyz")); StramLocalCluster lc = new StramLocalCluster(dag); StreamingContainerManager dnmgr = lc.dnmgr; StramAppContext appContext = new StramTestSupport.TestAppContext(dag.getAttributes()); AppDataPushAgent pushAgent = new AppDataPushAgent(dnmgr, appContext); pushAgent.init(); pushAgent.pushData(); Assert.assertTrue(TestMetricTransport.messages.size() > 0); pushAgent.close(); String msg = TestMetricTransport.messages.get(0); Assert.assertTrue(msg.startsWith("xyz:")); }