You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2016/08/22 08:53:13 UTC
[1/2] flink git commit: [hotfix] [metrics] Rename host format constant
Repository: flink
Updated Branches:
refs/heads/master 59eb4332f -> ebda2e0ea
[hotfix] [metrics] Rename host format constant
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ebda2e0e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ebda2e0e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ebda2e0e
Branch: refs/heads/master
Commit: ebda2e0ea05a8363db910c5bd03b0b7cfa5ae279
Parents: cbbd800
Author: zentol <ch...@apache.org>
Authored: Wed Aug 17 15:05:45 2016 +0200
Committer: zentol <ch...@apache.org>
Committed: Mon Aug 22 10:52:55 2016 +0200
----------------------------------------------------------------------
.../flink/runtime/metrics/groups/JobManagerMetricGroup.java | 2 +-
.../flink/runtime/metrics/groups/TaskManagerMetricGroup.java | 2 +-
.../flink/runtime/metrics/scope/JobManagerJobScopeFormat.java | 2 +-
.../flink/runtime/metrics/scope/JobManagerScopeFormat.java | 2 +-
.../flink/runtime/metrics/scope/OperatorScopeFormat.java | 4 ++--
.../org/apache/flink/runtime/metrics/scope/ScopeFormat.java | 6 +++---
.../flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java | 2 +-
.../flink/runtime/metrics/scope/TaskManagerScopeFormat.java | 2 +-
.../apache/flink/runtime/metrics/scope/TaskScopeFormat.java | 4 ++--
.../apache/flink/runtime/metrics/groups/OperatorGroupTest.java | 2 +-
10 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
index 4aee742..da0d8f8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
@@ -93,7 +93,7 @@ public class JobManagerMetricGroup extends ComponentMetricGroup<JobManagerMetric
@Override
protected void putVariables(Map<String, String> variables) {
- variables.put(ScopeFormat.SCOPE_ACTOR_HOST, hostname);
+ variables.put(ScopeFormat.SCOPE_HOST, hostname);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
index fb1ab8d..8f81cfd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
@@ -118,7 +118,7 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup<TaskManagerMetr
@Override
protected void putVariables(Map<String, String> variables) {
- variables.put(ScopeFormat.SCOPE_ACTOR_HOST, hostname);
+ variables.put(ScopeFormat.SCOPE_HOST, hostname);
variables.put(ScopeFormat.SCOPE_TASKMANAGER_ID, taskManagerId);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerJobScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerJobScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerJobScopeFormat.java
index 2d5356a..d8860d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerJobScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerJobScopeFormat.java
@@ -28,7 +28,7 @@ public class JobManagerJobScopeFormat extends ScopeFormat {
public JobManagerJobScopeFormat(String format, JobManagerScopeFormat parentFormat) {
super(format, parentFormat, new String[] {
- SCOPE_ACTOR_HOST,
+ SCOPE_HOST,
SCOPE_JOB_ID,
SCOPE_JOB_NAME
});
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerScopeFormat.java
index 14f1b72..dbf412b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/JobManagerScopeFormat.java
@@ -25,7 +25,7 @@ public class JobManagerScopeFormat extends ScopeFormat {
public JobManagerScopeFormat(String format) {
super(format, null, new String[] {
- SCOPE_ACTOR_HOST
+ SCOPE_HOST
});
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/OperatorScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/OperatorScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/OperatorScopeFormat.java
index fcebe37..f07b807 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/OperatorScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/OperatorScopeFormat.java
@@ -27,7 +27,7 @@ public class OperatorScopeFormat extends ScopeFormat {
public OperatorScopeFormat(String format, TaskScopeFormat parentFormat) {
super(format, parentFormat, new String[] {
- SCOPE_ACTOR_HOST,
+ SCOPE_HOST,
SCOPE_TASKMANAGER_ID,
SCOPE_JOB_ID,
SCOPE_JOB_NAME,
@@ -57,4 +57,4 @@ public class OperatorScopeFormat extends ScopeFormat {
};
return bindVariables(template, values);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/ScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/ScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/ScopeFormat.java
index a45df41..adf0a85 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/ScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/ScopeFormat.java
@@ -68,13 +68,13 @@ public abstract class ScopeFormat {
// Scope Variables
// ------------------------------------------------------------------------
- public static final String SCOPE_ACTOR_HOST = asVariable("host");
+ public static final String SCOPE_HOST = asVariable("host");
// ----- Job Manager ----
/** The default scope format of the JobManager component: {@code "<host>.jobmanager"} */
public static final String DEFAULT_SCOPE_JOBMANAGER_COMPONENT =
- concat(SCOPE_ACTOR_HOST, "jobmanager");
+ concat(SCOPE_HOST, "jobmanager");
/** The default scope format of JobManager metrics: {@code "<host>.jobmanager"} */
public static final String DEFAULT_SCOPE_JOBMANAGER_GROUP = DEFAULT_SCOPE_JOBMANAGER_COMPONENT;
@@ -85,7 +85,7 @@ public abstract class ScopeFormat {
/** The default scope format of the TaskManager component: {@code "<host>.taskmanager.<tm_id>"} */
public static final String DEFAULT_SCOPE_TASKMANAGER_COMPONENT =
- concat(SCOPE_ACTOR_HOST, "taskmanager", SCOPE_TASKMANAGER_ID);
+ concat(SCOPE_HOST, "taskmanager", SCOPE_TASKMANAGER_ID);
/** The default scope format of TaskManager metrics: {@code "<host>.taskmanager.<tm_id>"} */
public static final String DEFAULT_SCOPE_TASKMANAGER_GROUP = DEFAULT_SCOPE_TASKMANAGER_COMPONENT;
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java
index 558e710..c33122f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerJobScopeFormat.java
@@ -28,7 +28,7 @@ public class TaskManagerJobScopeFormat extends ScopeFormat {
public TaskManagerJobScopeFormat(String format, TaskManagerScopeFormat parentFormat) {
super(format, parentFormat, new String[] {
- SCOPE_ACTOR_HOST,
+ SCOPE_HOST,
SCOPE_TASKMANAGER_ID,
SCOPE_JOB_ID,
SCOPE_JOB_NAME
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerScopeFormat.java
index 6e7c39d..23166a9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskManagerScopeFormat.java
@@ -25,7 +25,7 @@ public class TaskManagerScopeFormat extends ScopeFormat {
public TaskManagerScopeFormat(String format) {
super(format, null, new String[] {
- SCOPE_ACTOR_HOST,
+ SCOPE_HOST,
SCOPE_TASKMANAGER_ID
});
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskScopeFormat.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskScopeFormat.java
index a781bf1..e9862e8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskScopeFormat.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/scope/TaskScopeFormat.java
@@ -28,7 +28,7 @@ public class TaskScopeFormat extends ScopeFormat {
public TaskScopeFormat(String format, TaskManagerJobScopeFormat parentFormat) {
super(format, parentFormat, new String[] {
- SCOPE_ACTOR_HOST,
+ SCOPE_HOST,
SCOPE_TASKMANAGER_ID,
SCOPE_JOB_ID,
SCOPE_JOB_NAME,
@@ -59,4 +59,4 @@ public class TaskScopeFormat extends ScopeFormat {
};
return bindVariables(template, values);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebda2e0e/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
index 9a97563..c193ac8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
@@ -72,7 +72,7 @@ public class OperatorGroupTest {
Map<String, String> variables = opGroup.getAllVariables();
- testVariable(variables, ScopeFormat.SCOPE_ACTOR_HOST, "theHostName");
+ testVariable(variables, ScopeFormat.SCOPE_HOST, "theHostName");
testVariable(variables, ScopeFormat.SCOPE_TASKMANAGER_ID, "test-tm-id");
testVariable(variables, ScopeFormat.SCOPE_JOB_ID, jid.toString());
testVariable(variables, ScopeFormat.SCOPE_JOB_NAME, "myJobName");
[2/2] flink git commit: [FLINK-4245] Expose all defined variables
Posted by ch...@apache.org.
[FLINK-4245] Expose all defined variables
This closes #2300.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/cbbd8003
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/cbbd8003
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/cbbd8003
Branch: refs/heads/master
Commit: cbbd80033204a922ccf148c5ac3f1be6de0d810d
Parents: 59eb433
Author: zentol <ch...@apache.org>
Authored: Wed Jul 27 11:25:27 2016 +0200
Committer: zentol <ch...@apache.org>
Committed: Mon Aug 22 10:52:55 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/metrics/MetricGroup.java | 10 +++++
.../groups/UnregisteredMetricsGroup.java | 8 ++++
.../metrics/groups/AbstractMetricGroup.java | 31 +++++++++++++--
.../metrics/groups/ComponentMetricGroup.java | 34 ++++++++++++++--
.../metrics/groups/GenericMetricGroup.java | 4 +-
.../groups/JobManagerJobMetricGroup.java | 10 +----
.../metrics/groups/JobManagerMetricGroup.java | 14 ++++++-
.../runtime/metrics/groups/JobMetricGroup.java | 19 ++++++++-
.../metrics/groups/OperatorMetricGroup.java | 18 ++++++---
.../metrics/groups/ProxyMetricGroup.java | 7 ++++
.../groups/TaskManagerJobMetricGroup.java | 8 +---
.../metrics/groups/TaskManagerMetricGroup.java | 11 +++++-
.../runtime/metrics/groups/TaskMetricGroup.java | 20 ++++++----
.../metrics/groups/AbstractMetricGroupTest.java | 41 ++++++++++++++++++++
.../runtime/metrics/groups/MetricGroupTest.java | 2 +-
.../metrics/groups/OperatorGroupTest.java | 41 ++++++++++++++++++++
16 files changed, 235 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
index 15f25b9..95387ae 100644
--- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
+++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
@@ -18,6 +18,8 @@
package org.apache.flink.metrics;
+import java.util.Map;
+
/**
* A MetricGroup is a named container for {@link Metric Metrics} and further metric subgroups.
*
@@ -142,6 +144,14 @@ public interface MetricGroup {
String[] getScopeComponents();
/**
+ * Returns a map of all variables and their associated value, for example
+ * {@code {"<host>"="host-7", "<tm_id>"="taskmanager-2"}}
+ *
+ * @return map of all variables and their associated value
+ */
+ Map<String, String> getAllVariables();
+
+ /**
* Returns the fully qualified metric name, for example
* {@code "host-7.taskmanager-2.window_word_count.my-mapper.metricName"}
*
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
index 1c7f0ce..3bbd0f6 100644
--- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
+++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
@@ -25,6 +25,9 @@ import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.SimpleCounter;
+import java.util.Collections;
+import java.util.Map;
+
/**
* A special {@link MetricGroup} that does not register any metrics at the metrics registry
* and any reporters.
@@ -87,6 +90,11 @@ public class UnregisteredMetricsGroup implements MetricGroup {
}
@Override
+ public Map<String, String> getAllVariables() {
+ return Collections.emptyMap();
+ }
+
+ @Override
public String getMetricIdentifier(String metricName) {
return metricName;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
index 3ac9966..ef2bc0c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -24,9 +24,8 @@ import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.metrics.SimpleCounter;
-
+import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,14 +52,22 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* from any metrics reporter and any internal maps. Note that even closed metrics groups
* return Counters, Gauges, etc to the code, to prevent exceptions in the monitored code.
* These metrics simply do not get reported any more, when created on a closed group.
+ *
+ * @param <A> The type of the parent MetricGroup
*/
-public abstract class AbstractMetricGroup implements MetricGroup {
+public abstract class AbstractMetricGroup<A extends AbstractMetricGroup<?>> implements MetricGroup {
/** shared logger */
private static final Logger LOG = LoggerFactory.getLogger(MetricGroup.class);
// ------------------------------------------------------------------------
+ /** The parent group containing this group */
+ protected final A parent;
+
+ /** The map containing all variables and their associated values, lazily computed. */
+ protected volatile Map<String, String> variables;
+
/** The registry that this metrics group belongs to */
protected final MetricRegistry registry;
@@ -83,9 +90,25 @@ public abstract class AbstractMetricGroup implements MetricGroup {
// ------------------------------------------------------------------------
- public AbstractMetricGroup(MetricRegistry registry, String[] scope) {
+ public AbstractMetricGroup(MetricRegistry registry, String[] scope, A parent) {
this.registry = checkNotNull(registry);
this.scopeComponents = checkNotNull(scope);
+ this.parent = parent;
+ }
+
+ public Map<String, String> getAllVariables() {
+ if (variables == null) { // avoid synchronization for common case
+ synchronized(this) {
+ if (variables == null) {
+ if (parent != null) {
+ variables = parent.getAllVariables();
+ } else { // this case should only be true for mock groups
+ variables = new HashMap<>();
+ }
+ }
+ }
+ }
+ return variables;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ComponentMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ComponentMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ComponentMetricGroup.java
index 7b4cb05..9a93fbd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ComponentMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ComponentMetricGroup.java
@@ -20,6 +20,9 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import java.util.HashMap;
+import java.util.Map;
+
/**
* Abstract {@link org.apache.flink.metrics.MetricGroup} for system components (e.g.,
* TaskManager, Job, Task, Operator).
@@ -32,8 +35,10 @@ import org.apache.flink.runtime.metrics.MetricRegistry;
* certain identifiers from the scope. The scope for metrics belonging to the "Task"
* group could for example include the task attempt number (more fine grained identification), or
* exclude it (for continuity of the namespace across failure and recovery).
+ *
+ * @param <P> The type of the parent MetricGroup.
*/
-public abstract class ComponentMetricGroup extends AbstractMetricGroup {
+public abstract class ComponentMetricGroup<P extends AbstractMetricGroup<?>> extends AbstractMetricGroup<P> {
/**
* Creates a new ComponentMetricGroup.
@@ -41,10 +46,33 @@ public abstract class ComponentMetricGroup extends AbstractMetricGroup {
* @param registry registry to register new metrics with
* @param scope the scope of the group
*/
- public ComponentMetricGroup(MetricRegistry registry, String[] scope) {
- super(registry, scope);
+ public ComponentMetricGroup(MetricRegistry registry, String[] scope, P parent) {
+ super(registry, scope, parent);
}
+ @Override
+ public Map<String, String> getAllVariables() {
+ if (variables == null) { // avoid synchronization for common case
+ synchronized (this) {
+ if (variables == null) {
+ variables = new HashMap<>();
+ putVariables(variables);
+ if (parent != null) { // not true for Job-/TaskManagerMetricGroup
+ variables.putAll(parent.getAllVariables());
+ }
+ }
+ }
+ }
+ return variables;
+ }
+
+ /**
+ * Enters all variables specific to this ComponentMetricGroup and their associated values into the map.
+ *
+ * @param variables map to enter variables and their values into
+ */
+ protected abstract void putVariables(Map<String, String> variables);
+
/**
* Closes the component group by removing and closing all metrics and subgroups
* (inherited from {@link AbstractMetricGroup}), plus closing and removing all dedicated
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
index 09b9ea5..569ad0f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
@@ -24,10 +24,10 @@ import org.apache.flink.runtime.metrics.MetricRegistry;
* A simple named {@link org.apache.flink.metrics.MetricGroup} that is used to hold
* subgroups of metrics.
*/
-public class GenericMetricGroup extends AbstractMetricGroup {
+public class GenericMetricGroup extends AbstractMetricGroup<AbstractMetricGroup<?>> {
public GenericMetricGroup(MetricRegistry registry, AbstractMetricGroup parent, String name) {
- super(registry, makeScopeComponents(parent, name));
+ super(registry, makeScopeComponents(parent, name), parent);
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerJobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerJobMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerJobMetricGroup.java
index 1396a5a..9d384d8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerJobMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerJobMetricGroup.java
@@ -29,19 +29,13 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* Special {@link org.apache.flink.metrics.MetricGroup} representing everything belonging to
* a specific job, running on the JobManager.
*/
-public class JobManagerJobMetricGroup extends JobMetricGroup {
-
- /** The metrics group that contains this group */
- private final JobManagerMetricGroup parent;
-
+public class JobManagerJobMetricGroup extends JobMetricGroup<JobManagerMetricGroup> {
public JobManagerJobMetricGroup(
MetricRegistry registry,
JobManagerMetricGroup parent,
JobID jobId,
@Nullable String jobName) {
- super(registry, jobId, jobName, registry.getScopeFormats().getJobManagerJobFormat().formatScope(checkNotNull(parent), jobId, jobName));
-
- this.parent = parent;
+ super(registry, checkNotNull(parent), jobId, jobName, registry.getScopeFormats().getJobManagerJobFormat().formatScope(checkNotNull(parent), jobId, jobName));
}
public final JobManagerMetricGroup parent() {
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
index 0b02c20..4aee742 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.HashMap;
import java.util.Map;
@@ -30,14 +31,14 @@ import java.util.Map;
* <p>Contains extra logic for adding jobs with tasks, and removing jobs when they do
* not contain tasks any more
*/
-public class JobManagerMetricGroup extends ComponentMetricGroup {
+public class JobManagerMetricGroup extends ComponentMetricGroup<JobManagerMetricGroup> {
private final Map<JobID, JobManagerJobMetricGroup> jobs = new HashMap<>();
private final String hostname;
public JobManagerMetricGroup(MetricRegistry registry, String hostname) {
- super(registry, registry.getScopeFormats().getJobManagerFormat().formatScope(hostname));
+ super(registry, registry.getScopeFormats().getJobManagerFormat().formatScope(hostname), null);
this.hostname = hostname;
}
@@ -86,6 +87,15 @@ public class JobManagerMetricGroup extends ComponentMetricGroup {
return jobs.size();
}
+ // ------------------------------------------------------------------------
+ // Component Metric Group Specifics
+ // ------------------------------------------------------------------------
+
+ @Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put(ScopeFormat.SCOPE_ACTOR_HOST, hostname);
+ }
+
@Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return jobs.values();
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
index 0c1c55c..e101d2f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
@@ -21,15 +21,19 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import javax.annotation.Nullable;
+import java.util.Map;
/**
* Special abstract {@link org.apache.flink.metrics.MetricGroup} representing everything belonging to
* a specific job.
+ *
+ * @param <C> The type of the parent ComponentMetricGroup.
*/
@Internal
-public abstract class JobMetricGroup extends ComponentMetricGroup {
+public abstract class JobMetricGroup<C extends ComponentMetricGroup<C>> extends ComponentMetricGroup<C> {
/** The ID of the job represented by this metrics group */
protected final JobID jobId;
@@ -42,10 +46,11 @@ public abstract class JobMetricGroup extends ComponentMetricGroup {
protected JobMetricGroup(
MetricRegistry registry,
+ C parent,
JobID jobId,
@Nullable String jobName,
String[] scope) {
- super(registry, scope);
+ super(registry, scope, parent);
this.jobId = jobId;
this.jobName = jobName;
@@ -59,4 +64,14 @@ public abstract class JobMetricGroup extends ComponentMetricGroup {
public String jobName() {
return jobName;
}
+
+ // ------------------------------------------------------------------------
+ // Component Metric Group Specifics
+ // ------------------------------------------------------------------------
+
+ @Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put(ScopeFormat.SCOPE_JOB_ID, jobId.toString());
+ variables.put(ScopeFormat.SCOPE_JOB_NAME, jobName);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
index bb01ac7..9352faf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
@@ -19,22 +19,22 @@
package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.Collections;
+import java.util.Map;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Special {@link org.apache.flink.metrics.MetricGroup} representing an Operator.
*/
-public class OperatorMetricGroup extends ComponentMetricGroup {
-
- /** The task metric group that contains this operator metric groups */
- private final TaskMetricGroup parent;
+public class OperatorMetricGroup extends ComponentMetricGroup<TaskMetricGroup> {
+ private final String operatorName;
public OperatorMetricGroup(MetricRegistry registry, TaskMetricGroup parent, String operatorName) {
- super(registry, registry.getScopeFormats().getOperatorFormat().formatScope(checkNotNull(parent), operatorName));
- this.parent = parent;
+ super(registry, registry.getScopeFormats().getOperatorFormat().formatScope(checkNotNull(parent), operatorName), parent);
+ this.operatorName = operatorName;
}
// ------------------------------------------------------------------------
@@ -48,6 +48,12 @@ public class OperatorMetricGroup extends ComponentMetricGroup {
// ------------------------------------------------------------------------
@Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put(ScopeFormat.SCOPE_OPERATOR_NAME, operatorName);
+ // we don't enter the subtask_index as the task group does that already
+ }
+
+ @Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return Collections.emptyList();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ProxyMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ProxyMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ProxyMetricGroup.java
index 6dd38a2..633c4af 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ProxyMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ProxyMetricGroup.java
@@ -24,6 +24,8 @@ import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.MetricGroup;
+import java.util.Map;
+
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -95,6 +97,11 @@ public class ProxyMetricGroup<P extends MetricGroup> implements MetricGroup {
}
@Override
+ public Map<String, String> getAllVariables() {
+ return parentMetricGroup.getAllVariables();
+ }
+
+ @Override
public String getMetricIdentifier(String metricName) {
return parentMetricGroup.getMetricIdentifier(metricName);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java
index 931c146..1f852c7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java
@@ -34,10 +34,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*
* <p>Contains extra logic for adding Tasks ({@link TaskMetricGroup}).
*/
-public class TaskManagerJobMetricGroup extends JobMetricGroup {
-
- /** The metrics group that contains this group */
- private final TaskManagerMetricGroup parent;
+public class TaskManagerJobMetricGroup extends JobMetricGroup<TaskManagerMetricGroup> {
/** Map from execution attempt ID (task identifier) to task metrics */
private final Map<AbstractID, TaskMetricGroup> tasks = new HashMap<>();
@@ -49,8 +46,7 @@ public class TaskManagerJobMetricGroup extends JobMetricGroup {
TaskManagerMetricGroup parent,
JobID jobId,
@Nullable String jobName) {
- super(registry, jobId, jobName, registry.getScopeFormats().getTaskManagerJobFormat().formatScope(checkNotNull(parent), jobId, jobName));
- this.parent = parent;
+ super(registry, parent, jobId, jobName, registry.getScopeFormats().getTaskManagerJobFormat().formatScope(checkNotNull(parent), jobId, jobName));
}
public final TaskManagerMetricGroup parent() {
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
index f4bd0c6..fb1ab8d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.HashMap;
import java.util.Map;
@@ -31,7 +32,7 @@ import java.util.Map;
* <p>Contains extra logic for adding jobs with tasks, and removing jobs when they do
* not contain tasks any more
*/
-public class TaskManagerMetricGroup extends ComponentMetricGroup {
+public class TaskManagerMetricGroup extends ComponentMetricGroup<TaskManagerMetricGroup> {
private final Map<JobID, TaskManagerJobMetricGroup> jobs = new HashMap<>();
@@ -41,7 +42,7 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup {
public TaskManagerMetricGroup(MetricRegistry registry, String hostname, String taskManagerId) {
- super(registry, registry.getScopeFormats().getTaskManagerFormat().formatScope(hostname, taskManagerId));
+ super(registry, registry.getScopeFormats().getTaskManagerFormat().formatScope(hostname, taskManagerId), null);
this.hostname = hostname;
this.taskManagerId = taskManagerId;
}
@@ -116,6 +117,12 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup {
// ------------------------------------------------------------------------
@Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put(ScopeFormat.SCOPE_ACTOR_HOST, hostname);
+ variables.put(ScopeFormat.SCOPE_TASKMANAGER_ID, taskManagerId);
+ }
+
+ @Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return jobs.values();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
index 00e680b..78fec97 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import org.apache.flink.util.AbstractID;
import javax.annotation.Nullable;
@@ -32,10 +33,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*
* <p>Contains extra logic for adding operators.
*/
-public class TaskMetricGroup extends ComponentMetricGroup {
-
- /** The job metrics group containing this task metrics group */
- private final TaskManagerJobMetricGroup parent;
+public class TaskMetricGroup extends ComponentMetricGroup<TaskManagerJobMetricGroup> {
private final Map<String, OperatorMetricGroup> operators = new HashMap<>();
@@ -65,10 +63,9 @@ public class TaskMetricGroup extends ComponentMetricGroup {
int subtaskIndex,
int attemptNumber) {
super(registry, registry.getScopeFormats().getTaskFormat().formatScope(
- checkNotNull(parent), vertexId, checkNotNull(executionId), taskName, subtaskIndex, attemptNumber));
+ checkNotNull(parent), vertexId, checkNotNull(executionId), taskName, subtaskIndex, attemptNumber), parent);
- this.parent = parent;
- this.executionId = executionId;
+ this.executionId = checkNotNull(executionId);
this.vertexId = vertexId;
this.taskName = taskName;
this.subtaskIndex = subtaskIndex;
@@ -148,6 +145,15 @@ public class TaskMetricGroup extends ComponentMetricGroup {
// ------------------------------------------------------------------------
@Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put(ScopeFormat.SCOPE_TASK_VERTEX_ID, vertexId.toString());
+ variables.put(ScopeFormat.SCOPE_TASK_NAME, taskName);
+ variables.put(ScopeFormat.SCOPE_TASK_ATTEMPT_ID, executionId.toString());
+ variables.put(ScopeFormat.SCOPE_TASK_ATTEMPT_NUM, String.valueOf(attemptNumber));
+ variables.put(ScopeFormat.SCOPE_TASK_SUBTASK_INDEX, String.valueOf(subtaskIndex));
+ }
+
+ @Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return operators.values();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
new file mode 100644
index 0000000..ddfbf4b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.metrics.groups;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+public class AbstractMetricGroupTest {
+ /**
+ * Verifies that no {@link NullPointerException} is thrown when {@link AbstractMetricGroup#getAllVariables()} is
+ * called and the parent is null.
+ */
+ @Test
+ public void testGetAllVariables() {
+ MetricRegistry registry = new MetricRegistry(new Configuration());
+
+ AbstractMetricGroup group = new AbstractMetricGroup<AbstractMetricGroup>(registry, new String[0], null) {
+ };
+ assertTrue(group.getAllVariables().isEmpty());
+
+ registry.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
index 16fb518..a27206d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
@@ -135,7 +135,7 @@ public class MetricGroupTest {
private static class DummyAbstractMetricGroup extends AbstractMetricGroup {
public DummyAbstractMetricGroup(MetricRegistry registry) {
- super(registry, new String[0]);
+ super(registry, new String[0], null);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/cbbd8003/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
index 4da713f..9a97563 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/OperatorGroupTest.java
@@ -21,12 +21,17 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import org.apache.flink.util.AbstractID;
import org.junit.Test;
+import java.util.Map;
+
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
public class OperatorGroupTest {
@@ -50,4 +55,40 @@ public class OperatorGroupTest {
registry.shutdown();
}
+
+ @Test
+ public void testVariables() {
+ MetricRegistry registry = new MetricRegistry(new Configuration());
+
+ JobID jid = new JobID();
+ AbstractID tid = new AbstractID();
+ AbstractID eid = new AbstractID();
+
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ TaskManagerJobMetricGroup jmGroup = new TaskManagerJobMetricGroup(registry, tmGroup, jid, "myJobName");
+ TaskMetricGroup taskGroup = new TaskMetricGroup(
+ registry, jmGroup, tid, eid, "aTaskName", 11, 0);
+ OperatorMetricGroup opGroup = new OperatorMetricGroup(registry, taskGroup, "myOpName");
+
+ Map<String, String> variables = opGroup.getAllVariables();
+
+ testVariable(variables, ScopeFormat.SCOPE_ACTOR_HOST, "theHostName");
+ testVariable(variables, ScopeFormat.SCOPE_TASKMANAGER_ID, "test-tm-id");
+ testVariable(variables, ScopeFormat.SCOPE_JOB_ID, jid.toString());
+ testVariable(variables, ScopeFormat.SCOPE_JOB_NAME, "myJobName");
+ testVariable(variables, ScopeFormat.SCOPE_TASK_VERTEX_ID, tid.toString());
+ testVariable(variables, ScopeFormat.SCOPE_TASK_NAME, "aTaskName");
+ testVariable(variables, ScopeFormat.SCOPE_TASK_ATTEMPT_ID, eid.toString());
+ testVariable(variables, ScopeFormat.SCOPE_TASK_SUBTASK_INDEX, "11");
+ testVariable(variables, ScopeFormat.SCOPE_TASK_ATTEMPT_NUM, "0");
+ testVariable(variables, ScopeFormat.SCOPE_OPERATOR_NAME, "myOpName");
+
+ registry.shutdown();
+ }
+
+ private static void testVariable(Map<String, String> variables, String key, String expectedValue) {
+ String actualValue = variables.get(key);
+ assertNotNull(actualValue);
+ assertEquals(expectedValue, actualValue);
+ }
}