SAMZA-2561: Add job features to MetricsHeader #1402

PawasChhokra · 2020-07-21T15:23:24Z

Issues: Currently, the MetricsHeader object emitted by the SamzaContainer does not exclude basic job level information.
Changes: Added a few features of the job to be emitted by the MetricsHeader
API Changes: With this change, the MetricsHeader class will emit other properties of the job like number of containers used, number of cores used, etc.
Upgrade instructions: None
Usage instructions: None
Tests: Modified existing tests to work with this change.

Sanil15

Thanks for the changes, Minor comments

Sanil15 · 2020-07-21T18:14:22Z

samza-core/src/main/java/org/apache/samza/metrics/ApiType.java

+
+package org.apache.samza.metrics;
+
+public enum ApiType {


nit: SamzaAPI not a better name?

Sanil15 · 2020-07-21T18:20:30Z

samza-core/src/main/java/org/apache/samza/util/Util.java

+  public static String getApiType(Config config) {
+    ApplicationConfig appConfig = new ApplicationConfig(config);
+    String appClass = appConfig.getAppClass();
+    if (appClass == null || appClass.isEmpty()) {


Is this ever true? for legacy task class applications?

Sanil15 · 2020-07-21T18:21:17Z

samza-core/src/main/java/org/apache/samza/util/Util.java

 import org.apache.samza.config.TaskConfig;
+import org.apache.samza.metrics.ApiType;
+import org.apache.samza.metrics.DeploymentType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;


 public class Util {


Not related to your change but JobInfoUtil is a better name for this Util class rather than just Util

Sanil15 · 2020-07-21T18:22:45Z

samza-core/src/test/java/org/apache/samza/diagnostics/TestDiagnosticsManager.java

@@ -59,6 +59,14 @@
  private int numPersistentStores = 2;
  private int containerNumCores = 2;
  private boolean autosizingEnabled = false;
+  private String deploymentType = "test deployment type";


consider adding valid values for deployType and API type, to unit test the util functions you just added

Sanil15 · 2020-07-21T18:23:55Z

samza-core/src/test/java/org/apache/samza/diagnostics/TestDiagnosticsManager.java

@@ -252,7 +264,17 @@ private void validateMetricsHeader(OutgoingMessageEnvelope outgoingMessageEnvelo
    Assert.assertEquals(metricsSnapshot.getHeader().getSamzaVersion(), samzaVersion);
    Assert.assertEquals(metricsSnapshot.getHeader().getHost(), hostname);
    Assert.assertEquals(metricsSnapshot.getHeader().getSource(), DiagnosticsManager.class.getName());
-


nit: revert whitespace change

rmatharu-zz · 2020-07-21T22:02:24Z

samza-core/src/main/java/org/apache/samza/metrics/DeploymentType.java

+/*
+ * 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.samza.metrics;
+
+public enum DeploymentType {
+  YARN, STANDALONE
+}


Single line enums like these can likely be added to the metrics-header class, since that seems like the only place theyre used no?
Same for API type above.

rmatharu-zz · 2020-07-21T23:41:30Z

samza-core/src/main/java/org/apache/samza/util/Util.java

+  public static int getContainerCount(Config config) {
+    JobConfig jobConfig = new JobConfig(config);
+    return jobConfig.getContainerCount();
+  }
+


Why even have it as a util?
Could the caller not cast the "config" object into JobConfig, TaskConfig and ClusterManagerConfig once and invoke the respective function, rather than do a cast in each separate util.

Same for getContainerMemoryMb, getNumCores, getThreadPoolSize, getSspGrouperFactory, getHostAffinityEnabled, getContainerRetryCount, getContainerRetryWindowMs, getMaxConcurrency.
Then none of these "util" methods need to be there.

rmatharu-zz · 2020-07-21T23:48:17Z

samza-core/src/main/java/org/apache/samza/util/Util.java

+  public static int getMaxJvmHeapMb() {
+    Long maxJvmHeapMb = Runtime.getRuntime().maxMemory() / (1024 * 1024);
+    return maxJvmHeapMb.intValue();
+  }


Can DiagnosticsUtil.buildDiagnosticsManager also invoke this util method, instead of calling Runtime.getRuntime().maxMemory directly ?

rmatharu-zz · 2020-07-21T23:49:31Z

samza-core/src/main/scala/org/apache/samza/diagnostics/DiagnosticsManager.java

-      Duration terminationDuration, boolean autosizingEnabled) {
+      Duration terminationDuration,
+      boolean autosizingEnabled,
+      String deploymentType,


Should deploymentType and api type be enums, since we defined an enum for them above?

rmatharu-zz · 2020-07-21T23:51:32Z

samza-core/src/main/scala/org/apache/samza/diagnostics/DiagnosticsStreamMessage.java

+            metricsSnapshot.getHeader().getContainerThreadPoolSize(), metricsSnapshot.getHeader().getHostAffinity(),
+            metricsSnapshot.getHeader().getSspGrouper(), metricsSnapshot.getHeader().getMaxContainerRetryCount(),
+            metricsSnapshot.getHeader().getContainerRetryWindowMs(), metricsSnapshot.getHeader().getTaskMaxConcurrency(),
+            metricsSnapshot.getHeader().getMaxJvmHeapMb());


Can we add a new separate constructor for DiagnosticsStreamMessage that takes a metricsHeader as input?

rmatharu-zz · 2020-07-21T23:52:49Z

...a-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala

+      serde,
+      blacklist,
+      Util.getDeploymentType(config),
+      Util.getApiType(config),
+      Util.getContainerCount(config),
+      Util.getContainerMemoryMb(config),
+      Util.getNumCores(config),
+      Util.getThreadPoolSize(config),
+      Util.getHostAffinityEnabled(config),
+      Util.getSspGrouperFactory(config),
+      Util.getContainerRetryCount(config),
+      Util.getContainerRetryWindowMs(config),
+      Util.getMaxConcurrency(config),
+      Util.getMaxJvmHeapMb)


Please maintain the existing ordering of passing in the parameters, then the serde and then the blacklist.
Similar for MetricsHeader above, where time fields are passed in last.

rmatharu-zz · 2020-07-21T23:53:03Z

samza-core/src/test/java/org/apache/samza/diagnostics/TestDiagnosticsManager.java

@@ -59,6 +59,14 @@
  private int numPersistentStores = 2;
  private int containerNumCores = 2;
  private boolean autosizingEnabled = false;
+  private String deploymentType = "test deployment type";
+  private String apiType = "test api type";


Wasnt this an enum ?

rmatharu-zz

Took an initial pass, requires some simplification and cleanup.

rmatharu-zz · 2020-07-24T18:53:08Z

Since a majority of the parameters are derived from config, in the interest of easy future extensibility, it'd be better to simply emit the entire config object once, from DiagnosticsManager at container-startup.
Similar for dependencies.

rmatharu-zz · 2020-07-30T18:12:16Z

samza-core/src/main/java/org/apache/samza/util/Util.java

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;


 public class Util {
  private static final Logger LOG = LoggerFactory.getLogger(Util.class);
+  private static final String YARN_JOB_FACTORY_CLASS = "org.apache.samza.job.yarn.YarnJobFactory";
+  private static final String BEAM_RUNNER_CLASS = "org.apache.beam.runners.samza.SamzaRunner";


Is this actually needed given
#1185 ?

rmatharu-zz

Comment about #1185

Add job parameters to MetricsHeader

f73b569

Sanil15 requested changes Jul 21, 2020

View reviewed changes

rmatharu-zz reviewed Jul 21, 2020

View reviewed changes

rmatharu-zz reviewed Jul 30, 2020

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SAMZA-2561: Add job features to MetricsHeader #1402

SAMZA-2561: Add job features to MetricsHeader #1402

PawasChhokra commented Jul 21, 2020

Sanil15 left a comment

Sanil15 Jul 21, 2020

Sanil15 Jul 21, 2020

Sanil15 Jul 21, 2020

Sanil15 Jul 21, 2020

Sanil15 Jul 21, 2020

rmatharu-zz Jul 21, 2020 •

edited

Loading

rmatharu-zz Jul 21, 2020

Sanil15 Jul 22, 2020

rmatharu-zz Jul 21, 2020

rmatharu-zz Jul 21, 2020

rmatharu-zz Jul 21, 2020

rmatharu-zz Jul 21, 2020

rmatharu-zz Jul 21, 2020

rmatharu-zz left a comment

rmatharu-zz commented Jul 24, 2020

rmatharu-zz Jul 30, 2020

rmatharu-zz left a comment

SAMZA-2561: Add job features to MetricsHeader #1402

Are you sure you want to change the base?

SAMZA-2561: Add job features to MetricsHeader #1402

Conversation

PawasChhokra commented Jul 21, 2020

Sanil15 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz Jul 21, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz left a comment

Choose a reason for hiding this comment

rmatharu-zz commented Jul 24, 2020

Choose a reason for hiding this comment

rmatharu-zz left a comment

Choose a reason for hiding this comment

rmatharu-zz Jul 21, 2020 •

edited

Loading