[
https://issues.apache.org/jira/browse/GOBBLIN-2104?focusedWorklogId=925681&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-925681
]
ASF GitHub Bot logged work on GOBBLIN-2104:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 12/Jul/24 17:48
Start Date: 12/Jul/24 17:48
Worklog Time Spent: 10m
Work Description: phet commented on code in PR #3983:
URL: https://github.com/apache/gobblin/pull/3983#discussion_r1676177633
##########
gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java:
##########
@@ -80,4 +80,27 @@ public class ServiceMetricNames {
public static final String DAG_COUNT_FS_DAG_STATE_COUNT =
GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "FsDagStateStore" + ".totalDagCount";
public static final String DAG_PROCESSING_EXCEPTION_METER =
"DagProcessingException";
+ /* DagProcessingEngine & Multi-active Execution Related Metrics
+ * Note: metrics ending with the delimiter '.' will be suffixed by the
specific {@link DagActionType} type for finer
+ * grained monitoring of each dagAction type in addition to the aggregation
of all types.
+ */
+ public static final String DAG_PROCESSING_ENGINE_PREFIX =
GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "DagProcEngine.";
+ public static final String DAG_ACTIONS_STORED = DAG_PROCESSING_ENGINE_PREFIX
+ "dagActionsStored.";
+ public static final String DAG_ACTIONS_OBSERVED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsObserved.";
+ public static final String DAG_ACTIONS_LEASES_OBTAINED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsLeasesObtained.";
+ public static final String DAG_ACTIONS_NO_LONGER_LEASING =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsNoLongerLeasing.";
+ public static final String DAG_ACTION_LEASE_REMINDERS_SCHEDULED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionLeaseRemindersScheduled.";
+ public static final String DAG_ACTION_REMINDERS_PROCESSED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionRemindersProcessed.";
+ // TODO: implement dropping reminder event after exceed some time
+ public static final String DAG_ACTIONS_EXCEEDED_MAX_RETRY =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsExceededMaxRetry.";
+ public static final String DAG_ACTIONS_INIT_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsInitFailed.";
+ public static final String DAG_ACTIONS_INIT_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsInitSucceeded.";
+ public static final String DAG_ACTION_EXECUTIONS_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionExecutionsFailed.";
+ public static final String DAG_ACTION_EXECUTIONS_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionExecutionsSucceeded.";
+ public static final String DAG_ACTION_CONCLUSIONS_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionConclusionsFailed.";
+ public static final String DAG_ACTION_CONCLUSIONS_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionConclusionsSucceeded.";
Review Comment:
for consistency in pluralization, I believe you want
`dagActionsReminderProcessed` and `dagActionsExecutionFailed`, etc.
but, that said, let's parallel method names, where they exist. so for
`DagProc::act`, `DagProc::initialize`, and `DagTask::conclude`, it would be
`dagActions{Initialize,Act,Conclude}{Succeeded,Failed}`
##########
gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java:
##########
@@ -80,4 +80,27 @@ public class ServiceMetricNames {
public static final String DAG_COUNT_FS_DAG_STATE_COUNT =
GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "FsDagStateStore" + ".totalDagCount";
public static final String DAG_PROCESSING_EXCEPTION_METER =
"DagProcessingException";
+ /* DagProcessingEngine & Multi-active Execution Related Metrics
+ * Note: metrics ending with the delimiter '.' will be suffixed by the
specific {@link DagActionType} type for finer
+ * grained monitoring of each dagAction type in addition to the aggregation
of all types.
+ */
+ public static final String DAG_PROCESSING_ENGINE_PREFIX =
GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "DagProcEngine.";
+ public static final String DAG_ACTIONS_STORED = DAG_PROCESSING_ENGINE_PREFIX
+ "dagActionsStored.";
+ public static final String DAG_ACTIONS_OBSERVED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsObserved.";
+ public static final String DAG_ACTIONS_LEASES_OBTAINED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsLeasesObtained.";
+ public static final String DAG_ACTIONS_NO_LONGER_LEASING =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsNoLongerLeasing.";
+ public static final String DAG_ACTION_LEASE_REMINDERS_SCHEDULED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionLeaseRemindersScheduled.";
+ public static final String DAG_ACTION_REMINDERS_PROCESSED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionRemindersProcessed.";
+ // TODO: implement dropping reminder event after exceed some time
+ public static final String DAG_ACTIONS_EXCEEDED_MAX_RETRY =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsExceededMaxRetry.";
+ public static final String DAG_ACTIONS_INIT_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsInitFailed.";
+ public static final String DAG_ACTIONS_INIT_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsInitSucceeded.";
+ public static final String DAG_ACTION_EXECUTIONS_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionExecutionsFailed.";
+ public static final String DAG_ACTION_EXECUTIONS_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionExecutionsSucceeded.";
+ public static final String DAG_ACTION_CONCLUSIONS_FAILED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionConclusionsFailed.";
+ public static final String DAG_ACTION_CONCLUSIONS_SUCCEEDED =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionConclusionsSucceeded.";
+ public static final String DAG_ACTIONS_REMOVED_FROM_STORE =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsRemovedFromStore.";
+ public static final String DAG_ACTIONS_FAILING_REMOVAL =
DAG_PROCESSING_ENGINE_PREFIX + "dagActionsFailingRemoval.";
Review Comment:
`dagActionsDelete{Succeeded,Failed}`?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java:
##########
@@ -64,8 +68,10 @@ protected void handleDagAction(String operation,
DagActionStore.DagAction dagAct
switch (operation) {
case "INSERT":
handleDagAction(dagAction, false);
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_OBSERVED,
dagActionType);
break;
case "UPDATE":
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_OBSERVED,
dagActionType);
Review Comment:
same concern here
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -229,16 +233,20 @@ protected void
processMessage(DecodeableKafkaRecord<String, DagActionStoreChange
dagActionsSeenCache.put(changeIdentifier, changeIdentifier);
}
- protected void handleDagAction(String operation, DagActionStore.DagAction
dagAction, String flowGroup, String flowName,
- long flowExecutionId, DagActionStore.DagActionType dagActionType) {
+ protected void handleDagAction(String operation, DagActionStore.DagAction
dagAction, String flowGroup,
+ String flowName, long flowExecutionId, DagActionStore.DagActionType
dagActionType) {
// We only expect INSERT and DELETE operations done to this table. INSERTs
correspond to any type of
// {@link DagActionStore.FlowActionType} flow requests that have to be
processed. DELETEs require no action.
try {
switch (operation) {
case "INSERT":
handleDagAction(dagAction, false);
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_OBSERVED,
+ dagActionType);
break;
case "UPDATE":
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_OBSERVED,
+ dagActionType);
Review Comment:
being as UPDATE is an error (that's already counted by its own metric),
should it really increment the same DPE metric as INSERT?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
+ * starting from addition to the DagActionStore to observation in the
DagActionChangeMonitor through the
+ * DagProcessingEngine pipeline (DagManagement -> DagTaskStreamImpl ->
MySqlMultiActiveLeaseArbiter -> DagProc).
+ */
+public class DagProcessingEngineMetrics {
+ MetricContext metricContext;
+ // Declare map of dagActionType to a ContextAwareMeter for each metric
+ private ConcurrentMap<String, ContextAwareMeter> dagActionsStoredMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsObservedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsLeasesObtainedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsNoLongerLeasingMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionLeaseRemindersScheduledMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionRemindersProcessedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsExceededMaxRetryMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsInitFailedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsInitSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsSucceededMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsRemovedFromStoreMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsFailingRemovalMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionAverageProcessingDelayMillisMeters = new
ConcurrentHashMap();
+
+ public DagProcessingEngineMetrics(MetricContext metricContext) {
+ this.metricContext = metricContext;
+ registerAllMetrics();
+ }
+
+ public void registerAllMetrics() {
+ registerMetricForEachDagAction(this.dagActionsStoredMeters,
ServiceMetricNames.DAG_ACTIONS_STORED);
+ registerMetricForEachDagAction(this.dagActionsObservedMeters,
ServiceMetricNames.DAG_ACTIONS_OBSERVED);
+ registerMetricForEachDagAction(this.dagActionsLeasesObtainedMeters,
ServiceMetricNames.DAG_ACTIONS_LEASES_OBTAINED);
+ registerMetricForEachDagAction(this.dagActionsNoLongerLeasingMeters,
ServiceMetricNames.DAG_ACTIONS_NO_LONGER_LEASING);
+
registerMetricForEachDagAction(this.dagActionLeaseRemindersScheduledMeters,
ServiceMetricNames.DAG_ACTION_LEASE_REMINDERS_SCHEDULED);
+ registerMetricForEachDagAction(this.dagActionRemindersProcessedMeters,
ServiceMetricNames.DAG_ACTION_REMINDERS_PROCESSED);
+ registerMetricForEachDagAction(this.dagActionsExceededMaxRetryMeters,
ServiceMetricNames.DAG_ACTIONS_EXCEEDED_MAX_RETRY);
+ registerMetricForEachDagAction(this.dagActionsInitFailedMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_FAILED);
+ registerMetricForEachDagAction(this.dagActionsInitSucceededMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionExecutionsFailedMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionExecutionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionConclusionsFailedMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionConclusionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_SUCCEEDED);
+ }
+
+ /**
+ * Create a meter of each dagActionType for the given metric, register it
with the metric context, and store it in a
+ * concurrent map.
+ * @param metricMap
+ * @param metricName
+ */
+ private void registerMetricForEachDagAction(ConcurrentMap<String,
ContextAwareMeter> metricMap, String metricName) {
+ for (DagActionStore.DagActionType dagActionType :
DagActionStore.DagActionType.values()) {
+ metricMap.put(dagActionType.toString(),
+ this.metricContext.contextAwareMeter(metricName + dagActionType));
+ }
+ }
+
+ /**
+ * Updates the meter corresponding to the metricsName and dagActionType
provided if they match an existing metric
+ * @param metricName
+ * @param dagActionType
+ */
+ public void updateMetricForDagAction(String metricName,
DagActionStore.DagActionType dagActionType) {
+ switch (metricName) {
+ case ServiceMetricNames.DAG_ACTIONS_STORED:
+ updateMetricForDagAction(this.dagActionsStoredMeters, dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTIONS_OBSERVED:
+ updateMetricForDagAction(this.dagActionsObservedMeters, dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTIONS_LEASES_OBTAINED:
+ updateMetricForDagAction(this.dagActionsLeasesObtainedMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTIONS_NO_LONGER_LEASING:
+ updateMetricForDagAction(this.dagActionsNoLongerLeasingMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_LEASE_REMINDERS_SCHEDULED:
+ updateMetricForDagAction(this.dagActionLeaseRemindersScheduledMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_REMINDERS_PROCESSED:
+ updateMetricForDagAction(this.dagActionRemindersProcessedMeters,
dagActionType);
+ break;
+ // TODO: implement evaluating max retries later
+ case ServiceMetricNames.DAG_ACTIONS_EXCEEDED_MAX_RETRY:
+ updateMetricForDagAction(this.dagActionsExceededMaxRetryMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTIONS_INIT_FAILED:
+ updateMetricForDagAction(this.dagActionsInitFailedMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTIONS_INIT_SUCCEEDED:
+ updateMetricForDagAction(this.dagActionsInitSucceededMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_EXECUTIONS_FAILED:
+ updateMetricForDagAction(this.dagActionExecutionsFailedMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_EXECUTIONS_SUCCEEDED:
+ updateMetricForDagAction(this.dagActionExecutionsSucceededMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_CONCLUSIONS_FAILED:
+ updateMetricForDagAction(this.dagActionConclusionsFailedMeters,
dagActionType);
+ break;
+ case ServiceMetricNames.DAG_ACTION_CONCLUSIONS_SUCCEEDED:
+ updateMetricForDagAction(this.dagActionConclusionsSucceededMeters,
dagActionType);
+ break;
+ default:
+ log.warn("Skipping marking metric because no meter found to match it.
Metric name: {}", metricName);
+ break;
+ }
+ }
+
+ /**
+ * Generic helper used to increment a metric corresponding to the
dagActionType in the provided map. It assumes the
+ * meter for each dagActionType can be identified by its name.
+ */
+ private void updateMetricForDagAction(ConcurrentMap<String,
ContextAwareMeter> metricMap,
+ DagActionStore.DagActionType dagActionType) {
+ if (metricMap.containsKey(dagActionType.toString())) {
+ metricMap.get(dagActionType.toString()).mark();
+ } else {
+ log.warn("Skipping metric. No meter exists for dagActionType {} in
metricsMap {}");
+ }
Review Comment:
this feels like a massive "can't happen" situation (given the strong typing
of `dagActionType`. therefore, it either ought to throw an exception or do a
JIT initialization of the `metricMap` for the new type
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
+ * starting from addition to the DagActionStore to observation in the
DagActionChangeMonitor through the
+ * DagProcessingEngine pipeline (DagManagement -> DagTaskStreamImpl ->
MySqlMultiActiveLeaseArbiter -> DagProc).
+ */
+public class DagProcessingEngineMetrics {
+ MetricContext metricContext;
+ // Declare map of dagActionType to a ContextAwareMeter for each metric
+ private ConcurrentMap<String, ContextAwareMeter> dagActionsStoredMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsObservedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsLeasesObtainedMeters = new ConcurrentHashMap();
Review Comment:
nit: all of these are missing a space between `>` and the var name. there
are also too many spaces between `= new`.
finally, naming them `XYZMeters` sounds like they're a list/set/sequence
collection, rather than a mapping.
hence, suggest: `leaseObtainedMeterByDagActionType` or
`initializeSucceededMeterByDagActionType`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -211,6 +221,11 @@ private LeaseAttemptStatus
retrieveLeaseStatus(DagActionStore.LeaseParams leaseP
*/
if (!(leaseAttemptStatus instanceof
LeaseAttemptStatus.NoLongerLeasingStatus)) {
scheduleReminderForEvent(leaseAttemptStatus);
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTION_LEASE_REMINDERS_SCHEDULED,
+ leaseParams.getDagAction().getDagActionType());
+ } else {
+
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_NO_LONGER_LEASING,
+ leaseParams.getDagAction().getDagActionType());
Review Comment:
gosh, everyone reaches in to pull out the same `DagActionType`... why not
encapsulate that within a `DagProcessingEngineMetrics` method taking either a
`LeaseParams` or an overall `DagAction`?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -132,11 +138,17 @@ public void run() {
DagProc<?> dagProc = dagTask.host(dagProcFactory);
try {
// todo - add retries
- dagProc.process(dagManagementStateStore);
+ dagProc.process(dagManagementStateStore, dagProcEngineMetrics);
dagTask.conclude();
+ // TODO: change this to determine the action type or make this a
generic meter
+
dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTION_CONCLUSIONS_SUCCEEDED,
+ DagActionStore.DagActionType.LAUNCH);
Review Comment:
could you put these two metrics calls within `DagTask::conclude` (since that
should have the `DagAction`)?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java:
##########
@@ -44,18 +46,28 @@ public KillDagProc(KillDagTask killDagTask) {
}
@Override
- protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore
dagManagementStateStore)
- throws IOException {
- return dagManagementStateStore.getDag(getDagId());
+ protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore
dagManagementStateStore,
+ DagProcessingEngineMetrics dagProcEngineMetrics) throws IOException {
+ try {
+ Optional<Dag<JobExecutionPlan>> dag =
dagManagementStateStore.getDag(getDagId());
+
dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_INIT_SUCCEEDED,
+ DagActionStore.DagActionType.KILL);
+ return dag;
+ } catch (Exception e) {
+
dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTIONS_INIT_FAILED,
+ DagActionStore.DagActionType.KILL);
+ throw e;
+ }
Review Comment:
this is far too much "bespoke" code in every one of the derived classes.
let's instead implement within the `DagProc` base class!
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -117,6 +122,7 @@ static class DagProcEngineThread implements Runnable {
private DagTaskStream dagTaskStream;
private DagProcFactory dagProcFactory;
private DagManagementStateStore dagManagementStateStore;
+ private DagProcessingEngineMetrics dagProcEngineMetrics;
Review Comment:
just wondering... would these ever change? why not `private final`?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
Review Comment:
did you mean "broken down by time" or "broken down by type"?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
+ * starting from addition to the DagActionStore to observation in the
DagActionChangeMonitor through the
+ * DagProcessingEngine pipeline (DagManagement -> DagTaskStreamImpl ->
MySqlMultiActiveLeaseArbiter -> DagProc).
+ */
Review Comment:
the convention I'm familiar w/ is to put javadoc first and then annotations
"closer" to the def. see examples from the official docs -
https://docs.oracle.com/javase/1.5.0/docs/guide/language/annotations.html
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProc.java:
##########
@@ -63,13 +64,16 @@ public DagProc(DagTask dagTask) {
this.dagNodeId = this.dagTask.getDagAction().getDagNodeId();
}
- public final void process(DagManagementStateStore dagManagementStateStore)
throws IOException {
- T state = initialize(dagManagementStateStore); // todo - retry
- act(dagManagementStateStore, state); // todo - retry
+ public final void process(DagManagementStateStore dagManagementStateStore,
+ DagProcessingEngineMetrics dagProcEngineMetrics) throws IOException {
+ T state = initialize(dagManagementStateStore, dagProcEngineMetrics); //
todo - retry
+ act(dagManagementStateStore, state, dagProcEngineMetrics); // todo -
retry
Review Comment:
these look like the two places to wrap the calls to increment metrics for
`{Initialize,Act}{Succeeded,Failed}`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
+ * starting from addition to the DagActionStore to observation in the
DagActionChangeMonitor through the
+ * DagProcessingEngine pipeline (DagManagement -> DagTaskStreamImpl ->
MySqlMultiActiveLeaseArbiter -> DagProc).
+ */
+public class DagProcessingEngineMetrics {
+ MetricContext metricContext;
+ // Declare map of dagActionType to a ContextAwareMeter for each metric
+ private ConcurrentMap<String, ContextAwareMeter> dagActionsStoredMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsObservedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsLeasesObtainedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsNoLongerLeasingMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionLeaseRemindersScheduledMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionRemindersProcessedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsExceededMaxRetryMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsInitFailedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsInitSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsSucceededMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsRemovedFromStoreMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsFailingRemovalMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionAverageProcessingDelayMillisMeters = new
ConcurrentHashMap();
+
+ public DagProcessingEngineMetrics(MetricContext metricContext) {
+ this.metricContext = metricContext;
+ registerAllMetrics();
+ }
+
+ public void registerAllMetrics() {
+ registerMetricForEachDagAction(this.dagActionsStoredMeters,
ServiceMetricNames.DAG_ACTIONS_STORED);
+ registerMetricForEachDagAction(this.dagActionsObservedMeters,
ServiceMetricNames.DAG_ACTIONS_OBSERVED);
+ registerMetricForEachDagAction(this.dagActionsLeasesObtainedMeters,
ServiceMetricNames.DAG_ACTIONS_LEASES_OBTAINED);
+ registerMetricForEachDagAction(this.dagActionsNoLongerLeasingMeters,
ServiceMetricNames.DAG_ACTIONS_NO_LONGER_LEASING);
+
registerMetricForEachDagAction(this.dagActionLeaseRemindersScheduledMeters,
ServiceMetricNames.DAG_ACTION_LEASE_REMINDERS_SCHEDULED);
+ registerMetricForEachDagAction(this.dagActionRemindersProcessedMeters,
ServiceMetricNames.DAG_ACTION_REMINDERS_PROCESSED);
+ registerMetricForEachDagAction(this.dagActionsExceededMaxRetryMeters,
ServiceMetricNames.DAG_ACTIONS_EXCEEDED_MAX_RETRY);
+ registerMetricForEachDagAction(this.dagActionsInitFailedMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_FAILED);
+ registerMetricForEachDagAction(this.dagActionsInitSucceededMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionExecutionsFailedMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionExecutionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionConclusionsFailedMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionConclusionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_SUCCEEDED);
+ }
+
+ /**
+ * Create a meter of each dagActionType for the given metric, register it
with the metric context, and store it in a
+ * concurrent map.
+ * @param metricMap
+ * @param metricName
+ */
+ private void registerMetricForEachDagAction(ConcurrentMap<String,
ContextAwareMeter> metricMap, String metricName) {
Review Comment:
nit: this is registering NOT `ForEachDagAction` but `ForEachDagActionType`
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java:
##########
@@ -80,17 +81,17 @@ public void setUp() throws Exception {
doReturn(true).when(dagActionStore).deleteDagAction(any());
dagManagementTaskStream =
new DagManagementTaskStreamImpl(config,
Optional.of(mock(DagActionStore.class)),
- mock(MultiActiveLeaseArbiter.class),
Optional.of(mock(DagActionReminderScheduler.class)), false,
- dagManagementStateStore);
+ mock(MultiActiveLeaseArbiter.class),
Optional.of(mock(DagActionReminderScheduler.class)),
+ false, dagManagementStateStore,
Mockito.mock(DagProcessingEngineMetrics.class));
this.dagProcFactory = new DagProcFactory(null);
DagProcessingEngine.DagProcEngineThread dagProcEngineThread =
new DagProcessingEngine.DagProcEngineThread(dagManagementTaskStream,
this.dagProcFactory,
- dagManagementStateStore, 0);
+ dagManagementStateStore, mock(DagProcessingEngineMetrics.class),
0);
Review Comment:
lots of tests are now being retrofitted to initialize w/ DPE metrics w/ a
mock.
is it not possible to verify behavior or what's invoked on that mock?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.orchestration.task;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.ConcurrentMap;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+
+@Slf4j
+/**
+ * Used to track all metrics relating to processing dagActions when
DagProcessingEngine is enabled. The metrics can be
+ * used to trace the number of dagActions (which can be further broken down by
time) at various points of the system,
+ * starting from addition to the DagActionStore to observation in the
DagActionChangeMonitor through the
+ * DagProcessingEngine pipeline (DagManagement -> DagTaskStreamImpl ->
MySqlMultiActiveLeaseArbiter -> DagProc).
+ */
+public class DagProcessingEngineMetrics {
+ MetricContext metricContext;
+ // Declare map of dagActionType to a ContextAwareMeter for each metric
+ private ConcurrentMap<String, ContextAwareMeter> dagActionsStoredMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsObservedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsLeasesObtainedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsNoLongerLeasingMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionLeaseRemindersScheduledMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionRemindersProcessedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsExceededMaxRetryMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String, ContextAwareMeter>dagActionsInitFailedMeters =
new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsInitSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionExecutionsSucceededMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsFailedMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionConclusionsSucceededMeters = new
ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsRemovedFromStoreMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionsFailingRemovalMeters = new ConcurrentHashMap();
+ private ConcurrentMap<String,
ContextAwareMeter>dagActionAverageProcessingDelayMillisMeters = new
ConcurrentHashMap();
+
+ public DagProcessingEngineMetrics(MetricContext metricContext) {
+ this.metricContext = metricContext;
+ registerAllMetrics();
+ }
+
+ public void registerAllMetrics() {
+ registerMetricForEachDagAction(this.dagActionsStoredMeters,
ServiceMetricNames.DAG_ACTIONS_STORED);
+ registerMetricForEachDagAction(this.dagActionsObservedMeters,
ServiceMetricNames.DAG_ACTIONS_OBSERVED);
+ registerMetricForEachDagAction(this.dagActionsLeasesObtainedMeters,
ServiceMetricNames.DAG_ACTIONS_LEASES_OBTAINED);
+ registerMetricForEachDagAction(this.dagActionsNoLongerLeasingMeters,
ServiceMetricNames.DAG_ACTIONS_NO_LONGER_LEASING);
+
registerMetricForEachDagAction(this.dagActionLeaseRemindersScheduledMeters,
ServiceMetricNames.DAG_ACTION_LEASE_REMINDERS_SCHEDULED);
+ registerMetricForEachDagAction(this.dagActionRemindersProcessedMeters,
ServiceMetricNames.DAG_ACTION_REMINDERS_PROCESSED);
+ registerMetricForEachDagAction(this.dagActionsExceededMaxRetryMeters,
ServiceMetricNames.DAG_ACTIONS_EXCEEDED_MAX_RETRY);
+ registerMetricForEachDagAction(this.dagActionsInitFailedMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_FAILED);
+ registerMetricForEachDagAction(this.dagActionsInitSucceededMeters,
ServiceMetricNames.DAG_ACTIONS_INIT_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionExecutionsFailedMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionExecutionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_EXECUTIONS_SUCCEEDED);
+ registerMetricForEachDagAction(this.dagActionConclusionsFailedMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_FAILED);
+ registerMetricForEachDagAction(this.dagActionConclusionsSucceededMeters,
ServiceMetricNames.DAG_ACTION_CONCLUSIONS_SUCCEEDED);
+ }
+
+ /**
+ * Create a meter of each dagActionType for the given metric, register it
with the metric context, and store it in a
+ * concurrent map.
+ * @param metricMap
+ * @param metricName
+ */
+ private void registerMetricForEachDagAction(ConcurrentMap<String,
ContextAwareMeter> metricMap, String metricName) {
+ for (DagActionStore.DagActionType dagActionType :
DagActionStore.DagActionType.values()) {
+ metricMap.put(dagActionType.toString(),
+ this.metricContext.contextAwareMeter(metricName + dagActionType));
+ }
+ }
+
+ /**
+ * Updates the meter corresponding to the metricsName and dagActionType
provided if they match an existing metric
+ * @param metricName
+ * @param dagActionType
+ */
+ public void updateMetricForDagAction(String metricName,
DagActionStore.DagActionType dagActionType) {
Review Comment:
a set of type-safe methods would be so preferable - a lot can go wrong w/
passing in `String metricName`!
e.g. instead of having a massive switch statement buried within the impl
which also pushes complexity (and the potential for errors) to callers, by
making them hard-code a reference to a well-known constant at each point of use:
```
this.dagProcEngineMetrics.updateMetricForDagAction(ServiceMetricNames.DAG_ACTION_REMINDERS_PROCESSED,
leaseParams.getDagAction());
```
instead ask them to hard-code a type-safe and semantic method name:
```
this.dagProcEngineMetrics.markReminderProcessed(leaseParams.getDagAction());
```
or
```
this.dagProcEngineMetrics.incrementInitializeSucceeded(dagAction);
```
the impl should be trivial:
```
public void markReminderProcessed(DagAction dagAction) {
updateMetric(this.reminderProcessedByDagActionType,
dagAction.getDagActionType());
}
```
Issue Time Tracking
-------------------
Worklog Id: (was: 925681)
Time Spent: 20m (was: 10m)
> Initialize DagProcessingEngine Metrics
> --------------------------------------
>
> Key: GOBBLIN-2104
> URL: https://issues.apache.org/jira/browse/GOBBLIN-2104
> Project: Apache Gobblin
> Issue Type: Bug
> Components: gobblin-service
> Reporter: Urmi Mustafi
> Assignee: Abhishek Tiwari
> Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
