MetricStore { @VisibleForTesting void add(MetricDump metric) { try { QueryScopeInfo info = metric.scopeInfo; TaskManagerMetricStore tm; JobMetricStore job; TaskMetricStore task; ComponentMetricStore subtask; String name = info.scope.isEmpty() ? metric.name : info.scope + "." + metric.name; if (name.isEmpty()) { return; } switch (info.getCategory()) { case INFO_CATEGORY_JM: addMetric(jobManager.metrics, name, metric); break; case INFO_CATEGORY_TM: String tmID = ((QueryScopeInfo.TaskManagerQueryScopeInfo) info).taskManagerID; tm = taskManagers.computeIfAbsent(tmID, k -> new TaskManagerMetricStore()); if (name.contains("GarbageCollector")) { String gcName = name.substring("Status.JVM.GarbageCollector.".length(), name.lastIndexOf('.')); tm.addGarbageCollectorName(gcName); } addMetric(tm.metrics, name, metric); break; case INFO_CATEGORY_JOB: QueryScopeInfo.JobQueryScopeInfo jobInfo = (QueryScopeInfo.JobQueryScopeInfo) info; job = jobs.computeIfAbsent(jobInfo.jobID, k -> new JobMetricStore()); addMetric(job.metrics, name, metric); break; case INFO_CATEGORY_TASK: QueryScopeInfo.TaskQueryScopeInfo taskInfo = (QueryScopeInfo.TaskQueryScopeInfo) info; job = jobs.computeIfAbsent(taskInfo.jobID, k -> new JobMetricStore()); task = job.tasks.computeIfAbsent(taskInfo.vertexID, k -> new TaskMetricStore()); subtask = task.subtasks.computeIfAbsent(taskInfo.subtaskIndex, k -> new ComponentMetricStore()); addMetric(subtask.metrics, name, metric); addMetric(task.metrics, taskInfo.subtaskIndex + "." + name, metric); break; case INFO_CATEGORY_OPERATOR: QueryScopeInfo.OperatorQueryScopeInfo operatorInfo = (QueryScopeInfo.OperatorQueryScopeInfo) info; job = jobs.computeIfAbsent(operatorInfo.jobID, k -> new JobMetricStore()); task = job.tasks.computeIfAbsent(operatorInfo.vertexID, k -> new TaskMetricStore()); subtask = task.subtasks.computeIfAbsent(operatorInfo.subtaskIndex, k -> new ComponentMetricStore()); addMetric(subtask.metrics, operatorInfo.operatorName + "." + name, metric); addMetric(task.metrics, operatorInfo.subtaskIndex + "." + operatorInfo.operatorName + "." + name, metric); break; default: LOG.debug("Invalid metric dump category: " + info.getCategory()); } } catch (Exception e) { LOG.debug("Malformed metric dump.", e); } }  synchronized ComponentMetricStore getJobManagerMetricStore(); synchronized TaskManagerMetricStore getTaskManagerMetricStore(String tmID); synchronized ComponentMetricStore getJobMetricStore(String jobID); synchronized TaskMetricStore getTaskMetricStore(String jobID, String taskID); synchronized ComponentMetricStore getSubtaskMetricStore(String jobID, String taskID, int subtaskIndex); synchronized Map<String, JobMetricStore> getJobs(); synchronized Map<String, TaskManagerMetricStore> getTaskManagers(); synchronized ComponentMetricStore getJobManager();  }
@Test public void testAdd() throws IOException { MetricStore store = setupStore(new MetricStore()); assertEquals("0", store.getJobManagerMetricStore().getMetric("abc.metric1", "-1")); assertEquals("1", store.getTaskManagerMetricStore("tmid").getMetric("abc.metric2", "-1")); assertEquals("2", store.getJobMetricStore("jobid").getMetric("abc.metric3", "-1")); assertEquals("3", store.getJobMetricStore("jobid").getMetric("abc.metric4", "-1")); assertEquals("4", store.getTaskMetricStore("jobid", "taskid").getMetric("8.abc.metric5", "-1")); assertEquals("5", store.getTaskMetricStore("jobid", "taskid").getMetric("8.opname.abc.metric6", "-1")); assertEquals("6", store.getTaskMetricStore("jobid", "taskid").getMetric("8.opname.abc.metric7", "-1")); }