You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2024/01/23 03:54:00 UTC
[jira] [Work logged] (GOBBLIN-1989) TemporaI: Implementation of GTE for GaaS Observability Event in MR deprecation for distcp
[ https://issues.apache.org/jira/browse/GOBBLIN-1989?focusedWorklogId=901099&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-901099 ]
ASF GitHub Bot logged work on GOBBLIN-1989:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 23/Jan/24 03:53
Start Date: 23/Jan/24 03:53
Worklog Time Spent: 10m
Work Description: phet commented on code in PR #3843:
URL: https://github.com/apache/gobblin/pull/3843#discussion_r1462608250
##########
gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/event/TimingEvent.java:
##########
@@ -162,7 +166,7 @@ public void doStop() {
}
this.stopped = true;
- this.endTime = System.currentTimeMillis();
+ this.endTime = getTime();
Review Comment:
for testing/mocking's sake it might be best not to call the same method for both start and end, but rather have a `getStartTime()` and `getEndTime()`
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/EventTimer.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.temporal.workflows.metrics;
+
+import java.io.Closeable;
+
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.event.EventSubmitter;
+import org.apache.gobblin.metrics.event.TimingEvent;
+
+
+/**
+ *
+ * A timer that can be used to track the duration of an event. This event differs from the {@link TimingEvent} in that
+ * this class is not meant to be used outside of {@link io.temporal.workflow.Workflow} code. We cannot use {@link TimingEvent}
+ * because it is not serializable and cannot be passed to {@link io.temporal.workflow.Workflow} code due to the
+ * EventSubmitter field. It also relies on `System.currentTimeMillis()` which not compatible with {@link io.temporal.workflow.Workflow}
+ * since {@link System#currentTimeMillis()} is not deterministic.
+ *
+ * The second main issue is {@link EventSubmitter} is not easily serializable because the {@link MetricContext} field
+ * contains bi-directional relationships via the {@link org.apache.gobblin.metrics.InnerGauge}. Although it's possible
+ * to write a custom serializer for {@link EventSubmitter}, it creates a non-obvious sleight of hand where the EventSubmitter
+ * metadata will change when crossing {@link io.temporal.workflow.Workflow} or {@link io.temporal.activity.Activity} boundaries.
+ *
+ * It differs from {@link Closeable} because the close method does not throw {@link java.io.IOException}. {@link TimingEvent}
+ * does this but it is not an interface
Review Comment:
was this sentence clipped?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -16,35 +16,86 @@
*/
package org.apache.gobblin.temporal.ddm.workflow.impl;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import org.apache.hadoop.fs.FileSystem;
import com.typesafe.config.ConfigFactory;
+import io.temporal.activity.ActivityOptions;
import io.temporal.api.enums.v1.ParentClosePolicy;
import io.temporal.workflow.ChildWorkflowOptions;
import io.temporal.workflow.Workflow;
import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.instrumented.GobblinMetricsKeys;
+import org.apache.gobblin.metrics.GobblinMetrics;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.util.JobMetrics;
import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.ddm.work.WUProcessingSpec;
import org.apache.gobblin.temporal.ddm.work.WorkUnitClaimCheck;
import org.apache.gobblin.temporal.ddm.work.assistance.Help;
import org.apache.gobblin.temporal.ddm.work.styles.FileSystemJobStateful;
import org.apache.gobblin.temporal.ddm.workflow.CommitStepWorkflow;
import org.apache.gobblin.temporal.ddm.workflow.ProcessWorkUnitsWorkflow;
-import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.util.nesting.work.WorkflowAddr;
import org.apache.gobblin.temporal.util.nesting.work.Workload;
import org.apache.gobblin.temporal.util.nesting.workflow.NestingExecWorkflow;
+import org.apache.gobblin.temporal.workflows.metrics.EventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.SubmitGTEActivity;
+import org.apache.gobblin.temporal.workflows.metrics.TemporalEventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.TrackingEventMetadata;
@Slf4j
public class ProcessWorkUnitsWorkflowImpl implements ProcessWorkUnitsWorkflow {
public static final String CHILD_WORKFLOW_ID_BASE = "NestingExecWorkUnits";
public static final String COMMIT_STEP_WORKFLOW_ID_BASE = "CommitStepWorkflow";
+ private final ActivityOptions options = ActivityOptions.newBuilder()
+ .setStartToCloseTimeout(Duration.ofSeconds(60))
+ .build();
+ private final SubmitGTEActivity submitGTEActivity = Workflow.newActivityStub(
+ SubmitGTEActivity.class, options);
+
+ private static final int MAX_DESERIALIZATION_FS_LOAD_ATTEMPTS = 5;
@Override
public int process(WUProcessingSpec workSpec) {
+ // NOTE: We are using the metrics tags from Job Props to create the metric context for the timer and NOT
+ // the deserialized jobState from HDFS that is created by the real distcp job. This is because the AZ runtime
+ // settings we want are for the job launcher that launched this Yarn job.
+
+ try {
+ FileSystem fs = Help.loadFileSystemForce(workSpec);
+ JobState jobState = Help.loadJobStateUncached(workSpec, fs);
+ List<Tag<?>> tagsFromCurrentJob = workSpec.getTags();
+ String metricsSuffix = workSpec.getMetricsSuffix();
+ List<Tag<?>> tags = getTags(tagsFromCurrentJob, metricsSuffix, jobState);
+
+ TemporalEventTimer.Factory timerFactory = new TemporalEventTimer.Factory(submitGTEActivity, new TrackingEventMetadata(tags, JobMetrics.NAMESPACE));
+ try (EventTimer timer = timerFactory.getJobTimer()) {
+ return performWork(workSpec);
Review Comment:
don't we need a way to pass info from `performWork()` into the GTE that's submitted/emitted?
e.g. we want to indicate whether the completion status is FAILED, CANCELED, COMPLETE? similarly, the GTE would convey the total size of successful copy entities.
for that would we need a method like `EventTimer::addMetadata` that could be called on `timer` (whether in the main flow or a `catch` block)?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -16,35 +16,86 @@
*/
package org.apache.gobblin.temporal.ddm.workflow.impl;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import org.apache.hadoop.fs.FileSystem;
import com.typesafe.config.ConfigFactory;
+import io.temporal.activity.ActivityOptions;
import io.temporal.api.enums.v1.ParentClosePolicy;
import io.temporal.workflow.ChildWorkflowOptions;
import io.temporal.workflow.Workflow;
import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.instrumented.GobblinMetricsKeys;
+import org.apache.gobblin.metrics.GobblinMetrics;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.util.JobMetrics;
import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.ddm.work.WUProcessingSpec;
import org.apache.gobblin.temporal.ddm.work.WorkUnitClaimCheck;
import org.apache.gobblin.temporal.ddm.work.assistance.Help;
import org.apache.gobblin.temporal.ddm.work.styles.FileSystemJobStateful;
import org.apache.gobblin.temporal.ddm.workflow.CommitStepWorkflow;
import org.apache.gobblin.temporal.ddm.workflow.ProcessWorkUnitsWorkflow;
-import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.util.nesting.work.WorkflowAddr;
import org.apache.gobblin.temporal.util.nesting.work.Workload;
import org.apache.gobblin.temporal.util.nesting.workflow.NestingExecWorkflow;
+import org.apache.gobblin.temporal.workflows.metrics.EventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.SubmitGTEActivity;
+import org.apache.gobblin.temporal.workflows.metrics.TemporalEventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.TrackingEventMetadata;
@Slf4j
public class ProcessWorkUnitsWorkflowImpl implements ProcessWorkUnitsWorkflow {
public static final String CHILD_WORKFLOW_ID_BASE = "NestingExecWorkUnits";
public static final String COMMIT_STEP_WORKFLOW_ID_BASE = "CommitStepWorkflow";
+ private final ActivityOptions options = ActivityOptions.newBuilder()
+ .setStartToCloseTimeout(Duration.ofSeconds(60))
+ .build();
+ private final SubmitGTEActivity submitGTEActivity = Workflow.newActivityStub(
+ SubmitGTEActivity.class, options);
+
+ private static final int MAX_DESERIALIZATION_FS_LOAD_ATTEMPTS = 5;
@Override
public int process(WUProcessingSpec workSpec) {
+ // NOTE: We are using the metrics tags from Job Props to create the metric context for the timer and NOT
+ // the deserialized jobState from HDFS that is created by the real distcp job. This is because the AZ runtime
+ // settings we want are for the job launcher that launched this Yarn job.
+
+ try {
+ FileSystem fs = Help.loadFileSystemForce(workSpec);
+ JobState jobState = Help.loadJobStateUncached(workSpec, fs);
+ List<Tag<?>> tagsFromCurrentJob = workSpec.getTags();
+ String metricsSuffix = workSpec.getMetricsSuffix();
+ List<Tag<?>> tags = getTags(tagsFromCurrentJob, metricsSuffix, jobState);
+
+ TemporalEventTimer.Factory timerFactory = new TemporalEventTimer.Factory(submitGTEActivity, new TrackingEventMetadata(tags, JobMetrics.NAMESPACE));
Review Comment:
seems clearer abstraction:
```
TrackingEventMetadata eventMetadata = getTrackingEventMetadata(workSpec, fs);
TemporalEventTimer.Factory timerFactory = new TemporalEventTimer.Factory(submitGTEActivity, eventMetadata);
```
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -16,35 +16,86 @@
*/
package org.apache.gobblin.temporal.ddm.workflow.impl;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import org.apache.hadoop.fs.FileSystem;
import com.typesafe.config.ConfigFactory;
+import io.temporal.activity.ActivityOptions;
import io.temporal.api.enums.v1.ParentClosePolicy;
import io.temporal.workflow.ChildWorkflowOptions;
import io.temporal.workflow.Workflow;
import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.instrumented.GobblinMetricsKeys;
+import org.apache.gobblin.metrics.GobblinMetrics;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.util.JobMetrics;
import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.ddm.work.WUProcessingSpec;
import org.apache.gobblin.temporal.ddm.work.WorkUnitClaimCheck;
import org.apache.gobblin.temporal.ddm.work.assistance.Help;
import org.apache.gobblin.temporal.ddm.work.styles.FileSystemJobStateful;
import org.apache.gobblin.temporal.ddm.workflow.CommitStepWorkflow;
import org.apache.gobblin.temporal.ddm.workflow.ProcessWorkUnitsWorkflow;
-import org.apache.gobblin.temporal.ddm.work.EagerFsDirBackedWorkUnitClaimCheckWorkload;
import org.apache.gobblin.temporal.util.nesting.work.WorkflowAddr;
import org.apache.gobblin.temporal.util.nesting.work.Workload;
import org.apache.gobblin.temporal.util.nesting.workflow.NestingExecWorkflow;
+import org.apache.gobblin.temporal.workflows.metrics.EventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.SubmitGTEActivity;
+import org.apache.gobblin.temporal.workflows.metrics.TemporalEventTimer;
+import org.apache.gobblin.temporal.workflows.metrics.TrackingEventMetadata;
@Slf4j
public class ProcessWorkUnitsWorkflowImpl implements ProcessWorkUnitsWorkflow {
public static final String CHILD_WORKFLOW_ID_BASE = "NestingExecWorkUnits";
public static final String COMMIT_STEP_WORKFLOW_ID_BASE = "CommitStepWorkflow";
+ private final ActivityOptions options = ActivityOptions.newBuilder()
+ .setStartToCloseTimeout(Duration.ofSeconds(60))
+ .build();
+ private final SubmitGTEActivity submitGTEActivity = Workflow.newActivityStub(
+ SubmitGTEActivity.class, options);
+
+ private static final int MAX_DESERIALIZATION_FS_LOAD_ATTEMPTS = 5;
@Override
public int process(WUProcessingSpec workSpec) {
+ // NOTE: We are using the metrics tags from Job Props to create the metric context for the timer and NOT
+ // the deserialized jobState from HDFS that is created by the real distcp job. This is because the AZ runtime
+ // settings we want are for the job launcher that launched this Yarn job.
+
+ try {
+ FileSystem fs = Help.loadFileSystemForce(workSpec);
+ JobState jobState = Help.loadJobStateUncached(workSpec, fs);
+ List<Tag<?>> tagsFromCurrentJob = workSpec.getTags();
+ String metricsSuffix = workSpec.getMetricsSuffix();
+ List<Tag<?>> tags = getTags(tagsFromCurrentJob, metricsSuffix, jobState);
+
+ TemporalEventTimer.Factory timerFactory = new TemporalEventTimer.Factory(submitGTEActivity, new TrackingEventMetadata(tags, JobMetrics.NAMESPACE));
+ try (EventTimer timer = timerFactory.getJobTimer()) {
+ return performWork(workSpec);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public int performWork(WUProcessingSpec workSpec) {
Review Comment:
why public?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/TemporalEventTimer.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.temporal.workflows.metrics;
+
+import java.time.Duration;
+import java.time.Instant;
+
+import io.temporal.workflow.Workflow;
+import lombok.RequiredArgsConstructor;
+
+import org.apache.gobblin.metrics.event.EventSubmitter;
+import org.apache.gobblin.metrics.event.GobblinEventBuilder;
+import org.apache.gobblin.metrics.event.TimingEvent;
+
+
+/**
+ * Boiler plate for tracking elapsed time of events that is compatible with {@link Workflow}
+ * by using activities to record time
+ *
+ * This class is very similar to {@link TimingEvent} but uses {@link Workflow} compatible APIs. It's possible to refactor
+ * this class to inherit the {@link TimingEvent} but extra care would be needed to remove the {@link EventSubmitter} field
+ * since that class is not serializable without losing some information
+ */
+@RequiredArgsConstructor
+public class TemporalEventTimer implements EventTimer {
+ private final SubmitGTEActivity trackingEventActivity;
+ private final GobblinEventBuilder eventBuilder;
+ private final TrackingEventMetadata trackingEventMetadata;
+ private final Instant startTime;
+
+ @Override
+ public void stop() {
+ stop(getCurrentTime());
+ }
+
+ private void stop(Instant endTime) {
+ this.eventBuilder.addMetadata(EventSubmitter.EVENT_TYPE, TimingEvent.METADATA_TIMING_EVENT);
+ this.eventBuilder.addMetadata(TimingEvent.METADATA_START_TIME, Long.toString(this.startTime.toEpochMilli()));
+ this.eventBuilder.addMetadata(TimingEvent.METADATA_END_TIME, Long.toString(endTime.toEpochMilli()));
+ Duration duration = Duration.between(this.startTime, endTime);
+ this.eventBuilder.addMetadata(TimingEvent.METADATA_DURATION, Long.toString(duration.toMillis()));
+
+ trackingEventActivity.submitGTE(this.eventBuilder, trackingEventMetadata);
+ }
+
+ private static Instant getCurrentTime() {
+ return Instant.ofEpochMilli(Workflow.currentTimeMillis());
+ }
+
+ public static class Factory {
+ private final SubmitGTEActivity submitGTEActivity;
+ private final TrackingEventMetadata trackingEventMetadata;
+
+ public Factory(SubmitGTEActivity submitGTEActivity, TrackingEventMetadata trackingEventMetadata) {
+ this.submitGTEActivity = submitGTEActivity;
+ this.trackingEventMetadata = trackingEventMetadata;
+ }
+
+ public TemporalEventTimer create(String eventName, Instant startTime) {
+ GobblinEventBuilder eventBuilder = new GobblinEventBuilder(eventName, trackingEventMetadata.getNamespace());
+ return new TemporalEventTimer(submitGTEActivity, eventBuilder, this.trackingEventMetadata, startTime);
+ }
+
+ public TemporalEventTimer create(String eventName) {
+ return create(eventName, getCurrentTime());
+ }
+
+ /**
+ * Creates a timer that emits separate events at the start and end of a job. This imitates the behavior in
+ * {@link org.apache.gobblin.runtime.AbstractJobLauncher} by using stubs to be compatible with the
+ * {@link org.apache.gobblin.runtime.job_monitor.KafkaAvroJobMonitor}
+ *
+ * @return a timer that emits an event at the beginning of the job and a completion event ends at the end of the job
+ */
+ public EventTimer getJobTimer() {
+ TemporalEventTimer startTimer = create(TimingEvent.LauncherTimings.JOB_START);
+ startTimer.stop(Instant.EPOCH); // Job start event contains a stub end time
+
+ return () -> {
+ TemporalEventTimer endTimer = create(TimingEvent.LauncherTimings.JOB_COMPLETE, startTimer.startTime);
+ endTimer.stop();
+ };
Review Comment:
how is this different than just:
```
return create(TimingEvent.LauncherTimings.JOB_COMPLETE, startTimer.startTime);
```
?
...may deserve a comment
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -85,4 +136,32 @@ protected CommitStepWorkflow createCommitStepWorkflow() {
return Workflow.newChildWorkflowStub(CommitStepWorkflow.class, childOpts);
}
+
+ private List<Tag<?>> getTags(List<Tag<?>> tagsFromCurJob, String metricsSuffix, JobState jobStateFromHdfs) {
+ // Construct new tags list by combining subset of tags on HDFS job state and the rest of the fields from the current job
+ Map<String, Tag<?>> tagsMap = new HashMap<>();
+ Set<String> tagKeysFromJobState = new HashSet<>(Arrays.asList(
+ TimingEvent.FlowEventConstants.FLOW_NAME_FIELD,
+ TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD,
+ TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ TimingEvent.FlowEventConstants.JOB_NAME_FIELD,
+ TimingEvent.FlowEventConstants.JOB_GROUP_FIELD));
+
+ // Step 1, Add tags from the AZ props using the original job (the one that launched this yarn app)
+ tagsFromCurJob.forEach(tag -> tagsMap.put(tag.getKey(), tag));
+
+ // Step 2. Add tags from the jobState (the original MR job on HDFS)
+ GobblinMetrics.getCustomTagsFromState(jobStateFromHdfs).stream()
+ .filter(tag -> tagKeysFromJobState.contains(tag.getKey()))
+ .forEach(tag -> tagsMap.put(tag.getKey(), tag));
+
+ // Step 2a (optional): Add a suffix to the FLOW_NAME_FIELD AND FLOW_GROUP_FIELDS to prevent collisions when testing
+ Consumer<String> addSuffix = (key) -> tagsMap.put(key, new Tag<>(key, tagsMap.get(key).getValue() + metricsSuffix));
+ addSuffix.accept(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD);
+ addSuffix.accept(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD);
Review Comment:
this mutation is harder to follow and lowers maintainability. can't we just make the adjustment while inserting the value in the first place?
```
GobblinMetrics.getCustomTagsFromState(jobStateFromHdfs).stream()
.filter(tag -> tagKeysFromJobState.contains(tag.getKey()))
.forEach(tag -> {
boolean isTargetKey = tag.getKey().equals(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD)
|| tag.getKey().equals(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD);
tagsMap.put(tag.getKey(), !isTargetKey ? tag : new Tag<>(tag.getKey(), tag.getValue() + metricsSuffix);
});
```
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/TrackingEventMetadata.java:
##########
@@ -0,0 +1,73 @@
+package org.apache.gobblin.temporal.workflows.metrics;
+
+import java.util.List;
+
+import com.codahale.metrics.Gauge;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Getter;
+
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.EventSubmitter;
+
+import static org.apache.gobblin.instrumented.GobblinMetricsKeys.CLASS_META;
+
+
+/**
+ * Package private impl for sending the core essence of an {@link EventSubmitter} over the wire
+ *
+ * TODO: Fix this description
+ *
+ * Serializer for jackson that does its best to serialize a {@link EventSubmitter} from a json representation.
+ * NOTE: Due to the nested bidirectional nature of the {@link org.apache.gobblin.metrics.InnerMetricContext} and
+ * usage of {@link Gauge} this is not a perfect serialization but it is sufficient for passing important monitoring
+ * like original Azkaban flow name and gobblin job id
+ */
+@Getter
+public class TrackingEventMetadata {
+ private final List<Tag<?>> tags;
+ private final String namespace;
+ private final Class callerClass;
+
+ @JsonCreator
+ private TrackingEventMetadata(
+ @JsonProperty("tags") List<Tag<?>> tags,
+ @JsonProperty("namespace") String namespace,
+ @JsonProperty("callerClass") Class callerClass) {
+ this.tags = tags;
+ this.namespace = namespace;
+ this.callerClass = callerClass;
+ }
+
+ public TrackingEventMetadata(List<Tag<?>> tags, String namespace) {
+ // Explicitly send class over the wire to avoid any classloader issues
+ this(tags, namespace, tags.stream()
+ .filter(tag -> tag.getKey().equals(CLASS_META))
+ .findAny()
+ .map(tag -> (String) tag.getValue())
+ .map(TrackingEventMetadata::resolveClass)
+ .orElse(TrackingEventMetadata.class));
+ }
+
+ public TrackingEventMetadata(EventSubmitter eventSubmitter) {
+ this(eventSubmitter.getTags(), eventSubmitter.getNamespace());
+ }
+
+ public EventSubmitter createEventSubmitter() {
+ MetricContext metricContext = Instrumented.getMetricContext(new State(), callerClass, tags);
+ return new EventSubmitter.Builder(metricContext, namespace).build();
+ }
Review Comment:
on the one hand, I like the encapsulation of magically turning one's TEM into an `EventSubmitter`... but it seems unintuitive and not memorable that "metadata" should double as an `EventSubmitter` factory.
I believe this is more a matter of naming than of an abstraction that's not cohesive and trying to do too many things. what's your take?
(perhaps the hang up for me is that events themselves contain metadata, such as the timing info you add to `EventBuilder`... but AFAICT this is NOT related to the extensible set of metadata that a particular activity might define to specify an event.)
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/TrackingEventMetadata.java:
##########
@@ -0,0 +1,73 @@
+package org.apache.gobblin.temporal.workflows.metrics;
+
+import java.util.List;
+
+import com.codahale.metrics.Gauge;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Getter;
+
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.EventSubmitter;
+
+import static org.apache.gobblin.instrumented.GobblinMetricsKeys.CLASS_META;
+
+
+/**
+ * Package private impl for sending the core essence of an {@link EventSubmitter} over the wire
+ *
+ * TODO: Fix this description
Review Comment:
can we complete before merging this PR?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java:
##########
@@ -43,6 +43,13 @@ public interface GobblinTemporalConfigurationKeys {
String GOBBLIN_TEMPORAL_JOB_LAUNCHER_ARG_PREFIX = GOBBLIN_TEMPORAL_JOB_LAUNCHER_PREFIX + "arg.";
+ /**
+ * Suffix for metrics emitted by GobblinTemporalJobLauncher for preventing collisions with prod jobs
+ * during testing
+ */
+ String GOBBLIN_TEMPORAL_JOB_METRICS_SUFFIX = PREFIX + "job.metrics.suffix";
+ String DEFAULT_GOBBLIN_TEMPORAL_JOB_METRICS_SUFFIX = "";
Review Comment:
still unclear on the answer here...
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/helloworld/GreetingWorkflowImpl.java:
##########
@@ -41,16 +50,20 @@ public class GreetingWorkflowImpl implements GreetingWorkflow {
*
* The activity options that were defined above are passed in as a parameter.
*/
- private final FormatActivity activity = Workflow.newActivityStub(FormatActivity.class, options);
+ private final FormatActivity formatActivity = Workflow.newActivityStub(FormatActivity.class, options);
+ private final SubmitGTEActivity
+ timerActivity = Workflow.newActivityStub(SubmitGTEActivity.class, options);
Review Comment:
aside from `options`, the `SubmitGTEActivityImpl` seems purely plumbing. given that, does it really make sense to require every caller to construct their own?
perhaps merely have the `TemporalEventTimer.Factory` constructor take in `ActivityOptions`, so the details of `SubmitGTEActivity` remain encapsulated. given the dynamic resolution by `Workflow.newActivityStub(SubmitGTEActivity.class, opts)`, I'm not sure there would ever even be the need to use a custom invocation. if so, that's an argument in favor of such encapsulation.
Issue Time Tracking
-------------------
Worklog Id: (was: 901099)
Remaining Estimate: 0h
Time Spent: 10m
> TemporaI: Implementation of GTE for GaaS Observability Event in MR deprecation for distcp
> -----------------------------------------------------------------------------------------
>
> Key: GOBBLIN-1989
> URL: https://issues.apache.org/jira/browse/GOBBLIN-1989
> Project: Apache Gobblin
> Issue Type: Improvement
> Reporter: Matthew Ho
> Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)