forked from datahub-project/datahub
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix(gms): filter out runs of a dataJob without any run-events (datahu…
- Loading branch information
1 parent
b33f23f
commit cd9b204
Showing
6 changed files
with
333 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
43 changes: 43 additions & 0 deletions
43
...src/main/java/com/linkedin/datahub/upgrade/config/BackfillDataProcessInstancesConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package com.linkedin.datahub.upgrade.config; | ||
|
|
||
| import com.linkedin.datahub.upgrade.system.NonBlockingSystemUpgrade; | ||
| import com.linkedin.datahub.upgrade.system.dataprocessinstances.BackfillDataProcessInstances; | ||
| import com.linkedin.metadata.entity.EntityService; | ||
| import com.linkedin.metadata.search.elasticsearch.ElasticSearchService; | ||
| import io.datahubproject.metadata.context.OperationContext; | ||
| import org.opensearch.client.RestHighLevelClient; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Conditional; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| @Configuration | ||
| @Conditional(SystemUpdateCondition.NonBlockingSystemUpdateCondition.class) | ||
| public class BackfillDataProcessInstancesConfig { | ||
|
|
||
| @Bean | ||
| public NonBlockingSystemUpgrade backfillProcessInstancesHasRunEvents( | ||
| final OperationContext opContext, | ||
| EntityService<?> entityService, | ||
| ElasticSearchService elasticSearchService, | ||
| RestHighLevelClient restHighLevelClient, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.enabled}") final boolean enabled, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.reprocess.enabled}") | ||
| boolean reprocessEnabled, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.batchSize}") final Integer batchSize, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.delayMs}") final Integer delayMs, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.totalDays}") Integer totalDays, | ||
| @Value("${systemUpdate.processInstanceHasRunEvents.windowDays}") Integer windowDays) { | ||
| return new BackfillDataProcessInstances( | ||
| opContext, | ||
| entityService, | ||
| elasticSearchService, | ||
| restHighLevelClient, | ||
| enabled, | ||
| reprocessEnabled, | ||
| batchSize, | ||
| delayMs, | ||
| totalDays, | ||
| windowDays); | ||
| } | ||
| } |
54 changes: 54 additions & 0 deletions
54
...om/linkedin/datahub/upgrade/system/dataprocessinstances/BackfillDataProcessInstances.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| package com.linkedin.datahub.upgrade.system.dataprocessinstances; | ||
|
|
||
| import com.google.common.collect.ImmutableList; | ||
| import com.linkedin.datahub.upgrade.UpgradeStep; | ||
| import com.linkedin.datahub.upgrade.system.NonBlockingSystemUpgrade; | ||
| import com.linkedin.metadata.entity.EntityService; | ||
| import com.linkedin.metadata.search.elasticsearch.ElasticSearchService; | ||
| import io.datahubproject.metadata.context.OperationContext; | ||
| import java.util.List; | ||
| import org.opensearch.client.RestHighLevelClient; | ||
|
|
||
| public class BackfillDataProcessInstances implements NonBlockingSystemUpgrade { | ||
|
|
||
| private final List<UpgradeStep> _steps; | ||
|
|
||
| public BackfillDataProcessInstances( | ||
| OperationContext opContext, | ||
| EntityService<?> entityService, | ||
| ElasticSearchService elasticSearchService, | ||
| RestHighLevelClient restHighLevelClient, | ||
| boolean enabled, | ||
| boolean reprocessEnabled, | ||
| Integer batchSize, | ||
| Integer batchDelayMs, | ||
| Integer totalDays, | ||
| Integer windowDays) { | ||
| if (enabled) { | ||
| _steps = | ||
| ImmutableList.of( | ||
| new BackfillDataProcessInstancesHasRunEventsStep( | ||
| opContext, | ||
| entityService, | ||
| elasticSearchService, | ||
| restHighLevelClient, | ||
| reprocessEnabled, | ||
| batchSize, | ||
| batchDelayMs, | ||
| totalDays, | ||
| windowDays)); | ||
| } else { | ||
| _steps = ImmutableList.of(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String id() { | ||
| return "BackfillDataProcessInstances"; | ||
| } | ||
|
|
||
| @Override | ||
| public List<UpgradeStep> steps() { | ||
| return _steps; | ||
| } | ||
| } |
213 changes: 213 additions & 0 deletions
213
...hub/upgrade/system/dataprocessinstances/BackfillDataProcessInstancesHasRunEventsStep.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,213 @@ | ||
| package com.linkedin.datahub.upgrade.system.dataprocessinstances; | ||
|
|
||
| import static com.linkedin.metadata.Constants.*; | ||
|
|
||
| import com.google.common.base.Throwables; | ||
| import com.linkedin.common.urn.Urn; | ||
| import com.linkedin.datahub.upgrade.UpgradeContext; | ||
| import com.linkedin.datahub.upgrade.UpgradeStep; | ||
| import com.linkedin.datahub.upgrade.UpgradeStepResult; | ||
| import com.linkedin.datahub.upgrade.impl.DefaultUpgradeStepResult; | ||
| import com.linkedin.metadata.boot.BootstrapStep; | ||
| import com.linkedin.metadata.entity.EntityService; | ||
| import com.linkedin.metadata.search.elasticsearch.ElasticSearchService; | ||
| import com.linkedin.metadata.utils.elasticsearch.IndexConvention; | ||
| import com.linkedin.upgrade.DataHubUpgradeState; | ||
| import io.datahubproject.metadata.context.OperationContext; | ||
| import java.io.IOException; | ||
| import java.net.URISyntaxException; | ||
| import java.time.Instant; | ||
| import java.time.temporal.ChronoUnit; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.function.Function; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.codehaus.jackson.node.JsonNodeFactory; | ||
| import org.codehaus.jackson.node.ObjectNode; | ||
| import org.opensearch.action.search.SearchRequest; | ||
| import org.opensearch.action.search.SearchResponse; | ||
| import org.opensearch.client.RequestOptions; | ||
| import org.opensearch.client.RestHighLevelClient; | ||
| import org.opensearch.index.query.QueryBuilder; | ||
| import org.opensearch.index.query.QueryBuilders; | ||
| import org.opensearch.search.aggregations.Aggregation; | ||
| import org.opensearch.search.aggregations.AggregationBuilders; | ||
| import org.opensearch.search.aggregations.bucket.composite.CompositeAggregation; | ||
| import org.opensearch.search.aggregations.bucket.composite.CompositeAggregationBuilder; | ||
| import org.opensearch.search.aggregations.bucket.composite.TermsValuesSourceBuilder; | ||
| import org.opensearch.search.builder.SearchSourceBuilder; | ||
|
|
||
| @Slf4j | ||
| public class BackfillDataProcessInstancesHasRunEventsStep implements UpgradeStep { | ||
|
|
||
| private static final String UPGRADE_ID = "BackfillDataProcessInstancesHasRunEvents"; | ||
| private static final Urn UPGRADE_ID_URN = BootstrapStep.getUpgradeUrn(UPGRADE_ID); | ||
|
|
||
| private final OperationContext opContext; | ||
| private final EntityService<?> entityService; | ||
| private final ElasticSearchService elasticSearchService; | ||
| private final RestHighLevelClient restHighLevelClient; | ||
|
|
||
| private final boolean reprocessEnabled; | ||
| private final Integer batchSize; | ||
| private final Integer batchDelayMs; | ||
|
|
||
| private final Integer totalDays; | ||
| private final Integer windowDays; | ||
|
|
||
| public BackfillDataProcessInstancesHasRunEventsStep( | ||
| OperationContext opContext, | ||
| EntityService<?> entityService, | ||
| ElasticSearchService elasticSearchService, | ||
| RestHighLevelClient restHighLevelClient, | ||
| boolean reprocessEnabled, | ||
| Integer batchSize, | ||
| Integer batchDelayMs, | ||
| Integer totalDays, | ||
| Integer windowDays) { | ||
| this.opContext = opContext; | ||
| this.entityService = entityService; | ||
| this.elasticSearchService = elasticSearchService; | ||
| this.restHighLevelClient = restHighLevelClient; | ||
| this.reprocessEnabled = reprocessEnabled; | ||
| this.batchSize = batchSize; | ||
| this.batchDelayMs = batchDelayMs; | ||
| this.totalDays = totalDays; | ||
| this.windowDays = windowDays; | ||
| } | ||
|
|
||
| @SuppressWarnings("BusyWait") | ||
| @Override | ||
| public Function<UpgradeContext, UpgradeStepResult> executable() { | ||
| return (context) -> { | ||
| TermsValuesSourceBuilder termsValuesSourceBuilder = | ||
| new TermsValuesSourceBuilder("urn").field("urn"); | ||
|
|
||
| ObjectNode json = JsonNodeFactory.instance.objectNode(); | ||
| json.put("hasRunEvents", true); | ||
|
|
||
| IndexConvention indexConvention = opContext.getSearchContext().getIndexConvention(); | ||
|
|
||
| String runEventsIndexName = | ||
| indexConvention.getTimeseriesAspectIndexName( | ||
| DATA_PROCESS_INSTANCE_ENTITY_NAME, DATA_PROCESS_INSTANCE_RUN_EVENT_ASPECT_NAME); | ||
|
|
||
| DataHubUpgradeState upgradeState = DataHubUpgradeState.SUCCEEDED; | ||
|
|
||
| Instant now = Instant.now(); | ||
| Instant overallStart = now.minus(totalDays, ChronoUnit.DAYS); | ||
| for (int i = 0; ; i++) { | ||
| Instant windowEnd = now.minus(i * windowDays, ChronoUnit.DAYS); | ||
| if (!windowEnd.isAfter(overallStart)) { | ||
| break; | ||
| } | ||
| Instant windowStart = windowEnd.minus(windowDays, ChronoUnit.DAYS); | ||
| if (windowStart.isBefore(overallStart)) { | ||
| // last iteration, cap at overallStart | ||
| windowStart = overallStart; | ||
| } | ||
|
|
||
| QueryBuilder queryBuilder = | ||
| QueryBuilders.boolQuery() | ||
| .must( | ||
| QueryBuilders.rangeQuery("@timestamp") | ||
| .gte(windowStart.toString()) | ||
| .lt(windowEnd.toString())); | ||
|
|
||
| CompositeAggregationBuilder aggregationBuilder = | ||
| AggregationBuilders.composite("aggs", List.of(termsValuesSourceBuilder)) | ||
| .size(batchSize); | ||
|
|
||
| while (true) { | ||
| SearchRequest searchRequest = new SearchRequest(runEventsIndexName); | ||
| searchRequest.source( | ||
| new SearchSourceBuilder() | ||
| .size(0) | ||
| .aggregation(aggregationBuilder) | ||
| .query(queryBuilder)); | ||
|
|
||
| SearchResponse response; | ||
|
|
||
| try { | ||
| response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); | ||
| } catch (IOException e) { | ||
| log.error(Throwables.getStackTraceAsString(e)); | ||
| log.error("Error querying index {}", runEventsIndexName); | ||
| upgradeState = DataHubUpgradeState.FAILED; | ||
| break; | ||
| } | ||
| List<Aggregation> aggregations = response.getAggregations().asList(); | ||
| if (aggregations.isEmpty()) { | ||
| break; | ||
| } | ||
| CompositeAggregation aggregation = (CompositeAggregation) aggregations.get(0); | ||
| Set<Urn> urns = new HashSet<>(); | ||
| for (CompositeAggregation.Bucket bucket : aggregation.getBuckets()) { | ||
| for (Object value : bucket.getKey().values()) { | ||
| try { | ||
| urns.add(Urn.createFromString(String.valueOf(value))); | ||
| } catch (URISyntaxException e) { | ||
| log.warn("Ignoring invalid urn {}", value); | ||
| } | ||
| } | ||
| } | ||
| if (!urns.isEmpty()) { | ||
| urns = entityService.exists(opContext, urns); | ||
| urns.forEach( | ||
| urn -> | ||
| elasticSearchService.upsertDocument( | ||
| opContext, | ||
| DATA_PROCESS_INSTANCE_ENTITY_NAME, | ||
| json.toString(), | ||
| indexConvention.getEntityDocumentId(urn))); | ||
| } | ||
| if (aggregation.afterKey() == null) { | ||
| break; | ||
| } | ||
| aggregationBuilder.aggregateAfter(aggregation.afterKey()); | ||
| if (batchDelayMs > 0) { | ||
| log.info("Sleeping for {} ms", batchDelayMs); | ||
| try { | ||
| Thread.sleep(batchDelayMs); | ||
| } catch (InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| BootstrapStep.setUpgradeResult(context.opContext(), UPGRADE_ID_URN, entityService); | ||
| return new DefaultUpgradeStepResult(id(), upgradeState); | ||
| }; | ||
| } | ||
|
|
||
| @Override | ||
| public String id() { | ||
| return UPGRADE_ID; | ||
| } | ||
|
|
||
| /** | ||
| * Returns whether the upgrade should proceed if the step fails after exceeding the maximum | ||
| * retries. | ||
| */ | ||
| @Override | ||
| public boolean isOptional() { | ||
| return true; | ||
| } | ||
|
|
||
| /** Returns whether the upgrade should be skipped. */ | ||
| @Override | ||
| public boolean skip(UpgradeContext context) { | ||
| if (reprocessEnabled) { | ||
| return false; | ||
| } | ||
|
|
||
| boolean previouslyRun = | ||
| entityService.exists( | ||
| context.opContext(), UPGRADE_ID_URN, DATA_HUB_UPGRADE_RESULT_ASPECT_NAME, true); | ||
| if (previouslyRun) { | ||
| log.info("{} was already run. Skipping.", id()); | ||
| } | ||
| return previouslyRun; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters