You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "adithya2754 (via GitHub)" <gi...@apache.org> on 2026/02/13 09:45:15 UTC
[PR] [GOBBLIN-2246] Aditrao/configurable mdm dimensions [gobblin]
adithya2754 opened a new pull request, #4164:
URL: https://github.com/apache/gobblin/pull/4164
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [GOBBLIN-2246](https://issues.apache.org/jira/browse/GOBBLIN-2246) issues and references them in the PR title.
- [GOBBLIN-2246](https://issues.apache.org/jira/browse/GOBBLIN-2246)
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
Dimension set used for MDM metric 'jobSucceeded' is hardcoded, this PR is to make the dimension set configurable.
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
Added unit tests
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2814931773
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
+ * @param state
+ * @return
+ */
+ private static Map<String, String> getConfiguredJobSucceededDimensionsMap(State state) {
+ String raw = state.getProp(JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, "");
+ if (StringUtils.isBlank(raw)) {
+ return null;
+ }
+ try {
+ Type type = new TypeToken<Map<String, String>>() {}.getType();
+ Map<String, String> parsed = GsonUtils.GSON.fromJson(raw, type);
+ if (parsed == null || parsed.isEmpty()) {
+ return null;
+ }
+ // Normalize into a deterministic insertion-order map.
+ return new LinkedHashMap<>(parsed);
+ } catch (Exception e) {
+ log.warn("Failed parsing jobSucceeded dimensionsMap config `{}`; falling back to hardcoded defaults. Raw value: {}",
+ JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, raw, e);
+ return null;
+ }
+ }
+
+ /**
+ * Adds an attribute to the OpenTelemetry event builder based on the mapping of MDM dimension key to GaaS observability event field key.
+ * @param builder
+ * @param mdmDimensionKey
+ * @param obsEventFieldKey
+ * @param event
+ */
+ private void addObsEventFieldAttribute(AttributesBuilder builder, String mdmDimensionKey, String obsEventFieldKey,
+ GaaSJobObservabilityEvent event) {
+ if (StringUtils.isBlank(mdmDimensionKey) || StringUtils.isBlank(obsEventFieldKey)) {
+ return;
+ }
+ String normalizedEventKey = obsEventFieldKey.trim();
+
+ Object value = getObsEventFieldValue(normalizedEventKey, event);
+ if (value == null) {
+ builder.put(mdmDimensionKey, DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE);
+ return;
+ }
+ if (value instanceof Number) {
+ builder.put(mdmDimensionKey, ((Number) value).longValue());
+ return;
+ }
Review Comment:
If we get a double, we will be sending a long value, is that intentional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2814944419
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
+ * @param state
+ * @return
+ */
+ private static Map<String, String> getConfiguredJobSucceededDimensionsMap(State state) {
+ String raw = state.getProp(JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, "");
+ if (StringUtils.isBlank(raw)) {
+ return null;
+ }
+ try {
+ Type type = new TypeToken<Map<String, String>>() {}.getType();
+ Map<String, String> parsed = GsonUtils.GSON.fromJson(raw, type);
+ if (parsed == null || parsed.isEmpty()) {
+ return null;
+ }
+ // Normalize into a deterministic insertion-order map.
+ return new LinkedHashMap<>(parsed);
+ } catch (Exception e) {
+ log.warn("Failed parsing jobSucceeded dimensionsMap config `{}`; falling back to hardcoded defaults. Raw value: {}",
+ JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, raw, e);
+ return null;
+ }
+ }
+
+ /**
+ * Adds an attribute to the OpenTelemetry event builder based on the mapping of MDM dimension key to GaaS observability event field key.
+ * @param builder
+ * @param mdmDimensionKey
+ * @param obsEventFieldKey
+ * @param event
+ */
+ private void addObsEventFieldAttribute(AttributesBuilder builder, String mdmDimensionKey, String obsEventFieldKey,
+ GaaSJobObservabilityEvent event) {
+ if (StringUtils.isBlank(mdmDimensionKey) || StringUtils.isBlank(obsEventFieldKey)) {
+ return;
+ }
+ String normalizedEventKey = obsEventFieldKey.trim();
+
+ Object value = getObsEventFieldValue(normalizedEventKey, event);
+ if (value == null) {
+ builder.put(mdmDimensionKey, DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE);
+ return;
+ }
+ if (value instanceof Number) {
+ builder.put(mdmDimensionKey, ((Number) value).longValue());
+ return;
+ }
Review Comment:
While I agree a dimension usually would not have floating values, but its better to call that out somewhere.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2814958483
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
Review Comment:
Can you check if this value can be easily passed from configs? Given that we haven't used such pattern yet, would be better to confirm once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2815097475
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
Review Comment:
Is this really required?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "adithya2754 (via GitHub)" <gi...@apache.org>.
adithya2754 commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2826068364
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
Review Comment:
not needed really, we can control at preaggregtion if we dont want to emit and anyway this will be controlled by us only
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
Review Comment:
we are using something similar for this `<entry key="metrics.reporting.opentelemetry.headers" value="{"Content-Type":"...`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
+ * @param state
+ * @return
+ */
+ private static Map<String, String> getConfiguredJobSucceededDimensionsMap(State state) {
+ String raw = state.getProp(JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, "");
+ if (StringUtils.isBlank(raw)) {
+ return null;
+ }
+ try {
+ Type type = new TypeToken<Map<String, String>>() {}.getType();
+ Map<String, String> parsed = GsonUtils.GSON.fromJson(raw, type);
+ if (parsed == null || parsed.isEmpty()) {
+ return null;
+ }
+ // Normalize into a deterministic insertion-order map.
+ return new LinkedHashMap<>(parsed);
+ } catch (Exception e) {
+ log.warn("Failed parsing jobSucceeded dimensionsMap config `{}`; falling back to hardcoded defaults. Raw value: {}",
+ JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, raw, e);
+ return null;
+ }
+ }
+
+ /**
+ * Adds an attribute to the OpenTelemetry event builder based on the mapping of MDM dimension key to GaaS observability event field key.
+ * @param builder
+ * @param mdmDimensionKey
+ * @param obsEventFieldKey
+ * @param event
+ */
+ private void addObsEventFieldAttribute(AttributesBuilder builder, String mdmDimensionKey, String obsEventFieldKey,
+ GaaSJobObservabilityEvent event) {
+ if (StringUtils.isBlank(mdmDimensionKey) || StringUtils.isBlank(obsEventFieldKey)) {
+ return;
+ }
+ String normalizedEventKey = obsEventFieldKey.trim();
+
+ Object value = getObsEventFieldValue(normalizedEventKey, event);
+ if (value == null) {
+ builder.put(mdmDimensionKey, DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE);
+ return;
+ }
+ if (value instanceof Number) {
+ builder.put(mdmDimensionKey, ((Number) value).longValue());
+ return;
+ }
Review Comment:
sure will add this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#issuecomment-3926118556
## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/4164?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 38.74%. Comparing base ([`6619cb1`](https://app.codecov.io/gh/apache/gobblin/commit/6619cb1253572523a9953d73859552a4086556e4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`7182596`](https://app.codecov.io/gh/apache/gobblin/commit/71825962c4cc830b21d6b2cdee19634e8f885517?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 4 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (6619cb1) and HEAD (7182596). Click for more details.
>
> <details><summary>HEAD has 2 uploads less than BASE</summary>
>
>| Flag | BASE (6619cb1) | HEAD (7182596) |
>|------|------|------|
>||3|1|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4164 +/- ##
=============================================
- Coverage 49.12% 38.74% -10.38%
+ Complexity 10253 1600 -8653
=============================================
Files 1924 389 -1535
Lines 75350 16030 -59320
Branches 8361 1590 -6771
=============================================
- Hits 37012 6211 -30801
+ Misses 35039 9319 -25720
+ Partials 3299 500 -2799
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/gobblin/pull/4164?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain merged PR #4164:
URL: https://github.com/apache/gobblin/pull/4164
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2246] Make Orchestrator MDM metric dimesnion configurable [gobblin]
Posted by "abhishekmjain (via GitHub)" <gi...@apache.org>.
abhishekmjain commented on code in PR #4164:
URL: https://github.com/apache/gobblin/pull/4164#discussion_r2814931773
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java:
##########
@@ -138,15 +192,261 @@ public void emitObservabilityEvent(final State jobState) {
this.eventCollector.add(event);
}
+ /**
+ * Creates dimensions for the OpenTelemetry event based on the configured mapping of MDM dimension keys to GaaS observability event field keys.
+ * If the config is not present or fails to parse, falls back to a default set of dimensions using hardcoded mappings.
+ * The dimension values are read from the GaaS observability event.
+ * If a value is missing or cannot be read, it defaults to "NA".
+ * @param event
+ * @return
+ */
public Attributes getEventAttributes(GaaSJobObservabilityEvent event) {
- Attributes tags = Attributes.builder().put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD, getOrDefault(event.getFlowName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD, getOrDefault(event.getFlowGroup(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD, getOrDefault(event.getJobName(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD, event.getFlowExecutionId())
- .put(TimingEvent.FlowEventConstants.SPEC_EXECUTOR_FIELD, getOrDefault(event.getExecutorId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .put(TimingEvent.FlowEventConstants.FLOW_EDGE_FIELD, getOrDefault(event.getFlowEdgeId(), DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE))
- .build();
- return tags;
+ int maxDimensions = getMaxJobSucceededDimensions();
+ Map<String, String> configuredMap = getConfiguredJobSucceededDimensionsMap(this.state);
+ if (configuredMap == null || configuredMap.isEmpty()) {
+ // Backward-compatible fallback: always emit the baseline dimensions.
+ // If the extra-dimensions gate is enabled, honor per-run dimensions from job properties.
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ addObsEventFieldAttribute(builder, entry.getKey(), entry.getValue(), event);
+ }
+ addExtraDimensionsFromJobProperties(builder, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP, event, maxDimensions);
+ return builder.build();
+ }
+
+ // Drop orchestrator dimensions that point to the same event field as a baseline dimension.
+ // This prevents capturing the same source field under multiple dimension keys.
+ Map<String, String> filteredConfiguredMap = dropDuplicateObsEventFieldMappings(configuredMap, JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+
+ // Ensure baseline dimensions are always emitted, even if orchestrator config is missing some keys.
+ Map<String, String> effectiveMap = new LinkedHashMap<>(filteredConfiguredMap);
+ for (Map.Entry<String, String> baselineEntry : JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.entrySet()) {
+ effectiveMap.putIfAbsent(baselineEntry.getKey(), baselineEntry.getValue());
+ }
+
+ // If orchestrator config would push us past the cap, drop orchestrator dimensions (keep baseline only).
+ if (effectiveMap.size() > maxDimensions) {
+ log.warn("jobSucceeded would emit {} baseline+orchestrator dimensions which exceeds maxDimensions={} (`{}`); dropping orchestrator dimensions",
+ effectiveMap.size(), maxDimensions, JOB_SUCCEEDED_MAX_DIMENSIONS_KEY);
+ effectiveMap = new LinkedHashMap<>(JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP);
+ }
+
+ AttributesBuilder builder = Attributes.builder();
+ for (Map.Entry<String, String> entry : effectiveMap.entrySet()) {
+ String mdmDimensionKey = entry.getKey();
+ String obsEventFieldKey = entry.getValue();
+ addObsEventFieldAttribute(builder, mdmDimensionKey, obsEventFieldKey, event);
+ }
+
+ addExtraDimensionsFromJobProperties(builder, effectiveMap, event, maxDimensions);
+
+ return builder.build();
+ }
+
+ private int getMaxJobSucceededDimensions() {
+ int jobSucceededMaxDimensions = DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS;
+ try {
+ jobSucceededMaxDimensions = this.state.getPropAsInt(JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS);
+ } catch (Exception e) {
+ log.warn("Invalid `{}` value; using default {}", JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, DEFAULT_JOB_SUCCEEDED_MAX_DIMENSIONS, e);
+ }
+ int baselineSize = JOB_SUCCEEDED_BASELINE_DIMENSIONS_MAP.size();
+ if (jobSucceededMaxDimensions < baselineSize) {
+ log.warn("Configured `{}`={} is less than baseline dimension count {}; using {} instead",
+ JOB_SUCCEEDED_MAX_DIMENSIONS_KEY, jobSucceededMaxDimensions, baselineSize, baselineSize);
+ return baselineSize;
+ }
+ return jobSucceededMaxDimensions;
+ }
+
+ private static Map<String, String> dropDuplicateObsEventFieldMappings(Map<String, String> configuredMap,
+ Map<String, String> baselineMap) {
+ Map<String, String> baselineObsFieldToKey = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : baselineMap.entrySet()) {
+ if (StringUtils.isNotBlank(entry.getValue())) {
+ baselineObsFieldToKey.put(entry.getValue().trim(), entry.getKey());
+ }
+ }
+
+ Map<String, String> seenObsFieldToKey = new LinkedHashMap<>();
+ Map<String, String> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : configuredMap.entrySet()) {
+ String dimensionKey = entry.getKey();
+ String obsFieldKey = entry.getValue();
+ if (StringUtils.isBlank(dimensionKey) || StringUtils.isBlank(obsFieldKey)) {
+ continue;
+ }
+ String normalizedObsFieldKey = obsFieldKey.trim();
+
+ String baselineKey = baselineObsFieldToKey.get(normalizedObsFieldKey);
+ if (baselineKey != null && !baselineKey.equals(dimensionKey)) {
+ log.warn("Ignoring jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under baseline dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, baselineKey);
+ continue;
+ }
+
+ String previousKey = seenObsFieldToKey.get(normalizedObsFieldKey);
+ if (previousKey != null && !previousKey.equals(dimensionKey)) {
+ log.warn("Ignoring duplicate jobSucceeded dimension mapping `{}` -> `{}` because field `{}` is already captured under dimension `{}`",
+ dimensionKey, obsFieldKey, normalizedObsFieldKey, previousKey);
+ continue;
+ }
+
+ seenObsFieldToKey.put(normalizedObsFieldKey, dimensionKey);
+ filtered.put(dimensionKey, obsFieldKey);
+ }
+ return filtered;
+ }
+
+ /**
+ * Reads the JSON string config for job succeeded dimensions map, which is expected to be in the format of
+ * {@code <mdm_dim_key>: <gaas_obs_event_field_key>}.
+ * @param state
+ * @return
+ */
+ private static Map<String, String> getConfiguredJobSucceededDimensionsMap(State state) {
+ String raw = state.getProp(JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, "");
+ if (StringUtils.isBlank(raw)) {
+ return null;
+ }
+ try {
+ Type type = new TypeToken<Map<String, String>>() {}.getType();
+ Map<String, String> parsed = GsonUtils.GSON.fromJson(raw, type);
+ if (parsed == null || parsed.isEmpty()) {
+ return null;
+ }
+ // Normalize into a deterministic insertion-order map.
+ return new LinkedHashMap<>(parsed);
+ } catch (Exception e) {
+ log.warn("Failed parsing jobSucceeded dimensionsMap config `{}`; falling back to hardcoded defaults. Raw value: {}",
+ JOB_SUCCEEDED_DIMENSIONS_MAP_KEY, raw, e);
+ return null;
+ }
+ }
+
+ /**
+ * Adds an attribute to the OpenTelemetry event builder based on the mapping of MDM dimension key to GaaS observability event field key.
+ * @param builder
+ * @param mdmDimensionKey
+ * @param obsEventFieldKey
+ * @param event
+ */
+ private void addObsEventFieldAttribute(AttributesBuilder builder, String mdmDimensionKey, String obsEventFieldKey,
+ GaaSJobObservabilityEvent event) {
+ if (StringUtils.isBlank(mdmDimensionKey) || StringUtils.isBlank(obsEventFieldKey)) {
+ return;
+ }
+ String normalizedEventKey = obsEventFieldKey.trim();
+
+ Object value = getObsEventFieldValue(normalizedEventKey, event);
+ if (value == null) {
+ builder.put(mdmDimensionKey, DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE);
+ return;
+ }
+ if (value instanceof Number) {
+ builder.put(mdmDimensionKey, ((Number) value).longValue());
+ return;
+ }
Review Comment:
If we get a double, we will be casting it to long, is that intentional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org