From fa90ef64cc21366a043066a14b0b9b9ce5de646f Mon Sep 17 00:00:00 2001 From: Ram Narayan Balaji Date: Fri, 17 Jul 2026 12:22:40 +0530 Subject: [PATCH] Add id-based workflow instance states endpoint with generic task-transition filter Adds GET /v1/governance/workflowInstanceStates/workflowInstanceId/{workflowInstanceId} returning the full stage-transition timeline for a workflow instance keyed by UUID. Why: - Existing name-based endpoint filters via entityFQNHash(workflowDefinitionName, instanceId). Renaming a WorkflowDefinition invalidates historical hashes and silently returns empty results. Id-based query bypasses the hash entirely by reading the workflowInstanceId column directly. - Task payloads carry workflowInstanceId (UUID). Callers should not have to resolve workflowDefinitionId -> name via a second round-trip, and should not break when a workflow definition is renamed. Filter: - Default response returns every stage the workflow entered (generic - works for any workflow definition). - Pass onlyTaskTransitions=true to narrow to stages driven by a user-task transition. Fingerprint: stage.variables contains _transitionId, written by user-task nodes when the user resolves the task. Applies to any workflow with task nodes; not DAR-specific. Existing name-based endpoint left unchanged for backwards compatibility. Tests: - Unit: WorkflowInstanceStateResourceTest (4 tests: default returns all, filter narrows, malformed rows filtered without crash, empty result). - IT: test_WorkflowInstanceStatesByInstanceIdEndpoint in WorkflowDefinitionResourceIT - spins up a tag-approval workflow, extracts the resulting task's workflowInstanceId, hits both default and filtered variants of the id-based endpoint, asserts default == name-based endpoint and filtered is a strict subset containing only stages with the _transitionId fingerprint. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../tests/WorkflowDefinitionResourceIT.java | 189 ++++++++++++++++++ .../WorkflowInstanceStateResource.java | 63 ++++++ .../WorkflowInstanceStateResourceTest.java | 184 +++++++++++++++++ 3 files changed, 436 insertions(+) create mode 100644 openmetadata-service/src/test/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResourceTest.java diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/WorkflowDefinitionResourceIT.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/WorkflowDefinitionResourceIT.java index ff9ad22bc33c..5701124c94ba 100644 --- a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/WorkflowDefinitionResourceIT.java +++ b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/WorkflowDefinitionResourceIT.java @@ -24,9 +24,11 @@ import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.function.BiConsumer; @@ -6553,6 +6555,193 @@ void test_reviewerChangeUpdatesApprovalTasks(TestNamespace ns) throws Exception "test_reviewerChangeUpdatesApprovalTasks completed successfully - task assignee successfully changed from reviewer1 to reviewer2"); } + @Test + @Order(44) + void test_WorkflowInstanceStatesByInstanceIdEndpoint(TestNamespace ns) throws Exception { + LOG.info("Starting test_WorkflowInstanceStatesByInstanceIdEndpoint"); + + OpenMetadataClient client = SdkClients.adminClient(); + ensureWorkflowEventConsumerIsActive(client); + + CreateUser createReviewer = + new CreateUser() + .withName(ns.prefix("wisReviewer")) + .withEmail(ns.prefix("wisReviewer") + "@example.com") + .withDisplayName("WIS Reviewer"); + User reviewer = client.users().create(createReviewer); + EntityReference reviewerRef = reviewer.getEntityReference(); + + String workflowName = ns.prefix("wisEndpointApprovalWorkflow"); + String approvalWorkflowJson = + """ + { + "name": "%s", + "displayName": "WIS Endpoint Approval Workflow", + "description": "Workflow used by test_WorkflowInstanceStatesByInstanceIdEndpoint", + "trigger": { + "type": "eventBasedEntity", + "config": { + "entityTypes": ["tag"], + "events": ["Created", "Updated"], + "exclude": ["reviewers"], + "filter": {} + }, + "output": ["relatedEntity", "updatedBy"] + }, + "nodes": [ + {"name": "start", "displayName": "Start", "type": "startEvent", "subType": "startEvent"}, + { + "name": "ApproveTag", + "displayName": "Approve Tag", + "type": "userTask", + "subType": "userApprovalTask", + "config": { + "assignees": {"addReviewers": true, "addOwners": false, "candidates": []}, + "approvalThreshold": 1, + "rejectionThreshold": 1 + }, + "input": ["relatedEntity"], + "inputNamespaceMap": {"relatedEntity": "global"}, + "output": ["updatedBy"], + "branches": ["true", "false"] + }, + {"name": "end", "displayName": "End", "type": "endEvent", "subType": "endEvent"} + ], + "edges": [ + {"from": "start", "to": "ApproveTag"}, + {"from": "ApproveTag", "to": "end", "condition": "true"}, + {"from": "ApproveTag", "to": "end", "condition": "false"} + ], + "config": {"storeStageStatus": true} + } + """; + + CreateWorkflowDefinition approvalWorkflow = + JsonUtils.readValue( + approvalWorkflowJson.formatted(workflowName), CreateWorkflowDefinition.class); + + WorkflowDefinition createdWorkflow = client.workflowDefinitions().create(approvalWorkflow); + trackWorkflow(createdWorkflow.getName(), createdWorkflow.getId().toString()); + waitForWorkflowDeployment(client, createdWorkflow.getName()); + + CreateClassification createClassification = + new CreateClassification() + .withName(ns.prefix("WisEndpointClassification")) + .withDescription("Classification for WIS endpoint test"); + Classification classification = client.classifications().create(createClassification); + + CreateTag createTag = + new CreateTag() + .withName("WisEndpointTag") + .withClassification(classification.getName()) + .withDescription("Tag used to spawn a workflow instance for the endpoint test") + .withReviewers(List.of(reviewerRef)); + Tag tag = client.tags().create(createTag); + + await() + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofSeconds(2)) + .until( + () -> !listOpenApprovalTasks(client, tag.getFullyQualifiedName()).getData().isEmpty()); + + Task approvalTask = + listOpenApprovalTasks(client, tag.getFullyQualifiedName()).getData().getFirst(); + UUID workflowInstanceId = approvalTask.getWorkflowInstanceId(); + assertNotNull( + workflowInstanceId, "Approval task must carry a workflowInstanceId to drive this test"); + + String statesBasePath = "/v1/governance/workflowInstanceStates"; + String idPath = statesBasePath + "/workflowInstanceId/" + workflowInstanceId; + RequestOptions rangeOptions = + RequestOptions.builder() + .queryParam("startTs", "0") + .queryParam("endTs", "9999999999999") + .build(); + String namePath = statesBasePath + "/" + createdWorkflow.getName() + "/" + workflowInstanceId; + + ObjectMapper mapper = + new ObjectMapper().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES); + + String allResponseJson = + client + .getHttpClient() + .executeForString(HttpMethod.GET, idPath, null, RequestOptions.builder().build()); + JsonNode allData = mapper.readTree(allResponseJson).get("data"); + assertNotNull(allData, "id-based endpoint must return a data array"); + assertTrue( + allData.isArray() && allData.size() > 0, + "id-based endpoint default response must include at least the user-task stage"); + + for (int i = 0; i < allData.size(); i++) { + JsonNode row = allData.get(i); + assertEquals( + workflowInstanceId.toString(), + row.get("workflowInstanceId").asText(), + "each state must be for the queried workflowInstanceId"); + JsonNode stage = row.get("stage"); + assertNotNull(stage, "state must carry a stage object"); + assertNotNull(stage.get("name"), "stage.name must be present"); + assertNotNull(stage.get("startedAt"), "stage.startedAt must be present"); + } + + long previousTs = Long.MIN_VALUE; + for (int i = 0; i < allData.size(); i++) { + long ts = allData.get(i).get("timestamp").asLong(); + assertTrue(ts >= previousTs, "states must be ordered by timestamp ascending"); + previousTs = ts; + } + + String filteredResponseJson = + client + .getHttpClient() + .executeForString( + HttpMethod.GET, + idPath, + null, + RequestOptions.builder().queryParam("onlyTaskTransitions", "true").build()); + JsonNode filteredData = mapper.readTree(filteredResponseJson).get("data"); + assertNotNull(filteredData, "filtered response must return a data array"); + assertTrue( + filteredData.size() > 0, + "filtered response should include at least the currently-open user-task stage"); + + for (int i = 0; i < filteredData.size(); i++) { + JsonNode stage = filteredData.get(i).get("stage"); + String stageName = stage.get("name").asText(); + JsonNode variables = stage.get("variables"); + assertNotNull(variables, "filtered rows must carry a variables map"); + assertTrue( + variables.has(stageName + "_transitionId"), + "onlyTaskTransitions=true must return only stages with a " + + "_transitionId variable (user-task fingerprint); offending stage: " + + stageName); + } + + Set allRowIds = collectIds(allData); + Set filteredRowIds = collectIds(filteredData); + assertTrue( + allRowIds.containsAll(filteredRowIds), + "default (all) response must be a superset of the filtered (user-task-only) response"); + + String nameResponseJson = + client.getHttpClient().executeForString(HttpMethod.GET, namePath, null, rangeOptions); + JsonNode nameData = mapper.readTree(nameResponseJson).get("data"); + assertNotNull(nameData, "name-based endpoint must return a data array"); + Set nameRowIds = collectIds(nameData); + assertEquals( + allRowIds, + nameRowIds, + "id-based default and name-based endpoint must return the same set of state rows"); + } + + private Set collectIds(JsonNode dataArray) { + Set ids = new HashSet<>(); + for (int i = 0; i < dataArray.size(); i++) { + ids.add(dataArray.get(i).get("id").asText()); + } + return ids; + } + @Test @Order(40) void test_ApiEndpointPeriodicBatchWorkflow(TestNamespace ns) throws JsonProcessingException { diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResource.java index 336aa0b85bac..db530c9420e4 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResource.java @@ -18,9 +18,12 @@ import jakarta.ws.rs.core.Context; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.SecurityContext; +import java.util.List; +import java.util.Map; import java.util.UUID; import lombok.NonNull; import lombok.extern.slf4j.Slf4j; +import org.openmetadata.schema.governance.workflows.Stage; import org.openmetadata.schema.governance.workflows.WorkflowInstanceState; import org.openmetadata.schema.type.MetadataOperation; import org.openmetadata.schema.utils.ResultList; @@ -176,6 +179,66 @@ public ResultList listForStateId( workflowDefinitionName, workflowInstanceId, offset, startTs, endTs, limitParam, latest); } + @GET + @Path("/workflowInstanceId/{workflowInstanceId}") + @Operation( + operationId = "getWorkflowInstanceStatesByWorkflowInstanceId", + summary = "Get all Workflow Instance States for a Workflow Instance id", + description = + "Get Workflow Instance States for a Workflow Instance id, ordered by timestamp " + + "ascending. Id-based lookup is stable across WorkflowDefinition renames because " + + "the underlying query filters on the workflowInstanceId column directly instead " + + "of an FQN-hash derived from the definition name. By default every stage the " + + "workflow entered is returned, including internal automation stages (policy " + + "enforcement, attribute updates, start/end events). Pass " + + "onlyTaskTransitions=true to narrow the response to stages driven by a " + + "task transition, identified generically by the presence of a " + + "_transitionId entry in the stage's variables map, which is written " + + "only by user-task nodes when a user resolves the task. Generic across any " + + "workflow that has task nodes.", + responses = { + @ApiResponse( + responseCode = "200", + description = "The Workflow Instance States", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = WorkflowInstanceStateResultList.class))) + }) + public ResultList listByWorkflowInstanceId( + @Context SecurityContext securityContext, + @Parameter(description = "Workflow Instance ID", schema = @Schema(type = "UUID")) + @PathParam("workflowInstanceId") + UUID workflowInstanceId, + @Parameter( + description = + "When true, return only stages driven by a task transition — i.e. rows whose " + + "variables map contains a _transitionId entry. Defaults to " + + "false, which returns every stage the workflow entered.", + schema = @Schema(type = "boolean")) + @DefaultValue("false") + @QueryParam("onlyTaskTransitions") + boolean onlyTaskTransitions) { + OperationContext operationContext = + new OperationContext(Entity.WORKFLOW_DEFINITION, MetadataOperation.VIEW_ALL); + ResourceContextInterface resourceContext = ReportDataContext.builder().build(); + authorizer.authorize(securityContext, operationContext, resourceContext); + List states = repository.listAllStatesForInstance(workflowInstanceId); + if (onlyTaskTransitions) { + states = states.stream().filter(WorkflowInstanceStateResource::isTaskTransition).toList(); + } + return new ResultList<>(states, null, null, states.size()); + } + + private static boolean isTaskTransition(WorkflowInstanceState state) { + Stage stage = state == null ? null : state.getStage(); + if (stage == null || stage.getName() == null) { + return false; + } + Map variables = stage.getVariables(); + return variables != null && variables.containsKey(stage.getName() + "_transitionId"); + } + @GET @Path("/{id}") @Operation( diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResourceTest.java new file mode 100644 index 000000000000..9ed72fa9d9ae --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/governance/WorkflowInstanceStateResourceTest.java @@ -0,0 +1,184 @@ +/* + * Copyright 2021 Collate + * Licensed 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.openmetadata.service.resources.governance; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import jakarta.ws.rs.core.SecurityContext; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.openmetadata.schema.governance.workflows.Stage; +import org.openmetadata.schema.governance.workflows.WorkflowInstanceState; +import org.openmetadata.schema.utils.ResultList; +import org.openmetadata.service.Entity; +import org.openmetadata.service.jdbi3.WorkflowInstanceStateRepository; +import org.openmetadata.service.security.Authorizer; + +class WorkflowInstanceStateResourceTest { + + private WorkflowInstanceStateResource createResource( + MockedStatic entityMock, + WorkflowInstanceStateRepository mockRepo, + Authorizer mockAuth) { + entityMock + .when(() -> Entity.getEntityClassFromType(Entity.WORKFLOW_INSTANCE_STATE)) + .thenReturn(WorkflowInstanceState.class); + entityMock + .when(() -> Entity.getEntityTimeSeriesRepository(Entity.WORKFLOW_INSTANCE_STATE)) + .thenReturn(mockRepo); + entityMock + .when(() -> Entity.registerTimeSeriesResourcePermissions(anyString())) + .thenAnswer(inv -> null); + return new WorkflowInstanceStateResource(mockAuth); + } + + private static WorkflowInstanceState state( + UUID instanceId, String stageName, Map variables) { + return new WorkflowInstanceState() + .withId(UUID.randomUUID()) + .withWorkflowInstanceId(instanceId) + .withStage(new Stage().withName(stageName).withStartedAt(1L).withVariables(variables)); + } + + @Test + void endpointReturnsAllStagesByDefault() { + try (MockedStatic entityMock = mockStatic(Entity.class)) { + WorkflowInstanceStateRepository mockRepo = mock(WorkflowInstanceStateRepository.class); + Authorizer mockAuth = mock(Authorizer.class); + WorkflowInstanceStateResource resource = createResource(entityMock, mockRepo, mockAuth); + + UUID workflowInstanceId = UUID.randomUUID(); + WorkflowInstanceState taskReview = + state( + workflowInstanceId, + "TaskReview", + Map.of("TaskReview_transitionId", "approve", "TaskReview_updatedBy", "ram.balaji")); + WorkflowInstanceState policyAgent = + state(workflowInstanceId, "PolicyAgent", Map.of("PolicyAgent_result", "manual")); + WorkflowInstanceState grantedAccess = + state( + workflowInstanceId, "GrantedAccess", Map.of("GrantedAccess_transitionId", "revoke")); + WorkflowInstanceState revokedEnd = + state(workflowInstanceId, "RevokedEnd", Map.of("RevokedEnd_stageInstanceStateId", "id")); + + when(mockRepo.listAllStatesForInstance(workflowInstanceId)) + .thenReturn(List.of(taskReview, policyAgent, grantedAccess, revokedEnd)); + + ResultList result = + resource.listByWorkflowInstanceId(mock(SecurityContext.class), workflowInstanceId, false); + + assertEquals( + 4, result.getData().size(), "default must return every stage the workflow entered"); + assertEquals(4, result.getPaging().getTotal()); + verify(mockRepo).listAllStatesForInstance(workflowInstanceId); + verify(mockAuth).authorize(any(), any(), any()); + } + } + + @Test + void endpointNarrowsToUserTaskStagesWhenRequested() { + try (MockedStatic entityMock = mockStatic(Entity.class)) { + WorkflowInstanceStateRepository mockRepo = mock(WorkflowInstanceStateRepository.class); + Authorizer mockAuth = mock(Authorizer.class); + WorkflowInstanceStateResource resource = createResource(entityMock, mockRepo, mockAuth); + + UUID workflowInstanceId = UUID.randomUUID(); + WorkflowInstanceState taskReview = + state( + workflowInstanceId, + "TaskReview", + Map.of("TaskReview_transitionId", "approve", "TaskReview_updatedBy", "ram.balaji")); + WorkflowInstanceState policyAgent = + state(workflowInstanceId, "PolicyAgent", Map.of("PolicyAgent_result", "manual")); + WorkflowInstanceState grantedAccess = + state( + workflowInstanceId, "GrantedAccess", Map.of("GrantedAccess_transitionId", "revoke")); + WorkflowInstanceState revokedEnd = + state(workflowInstanceId, "RevokedEnd", Map.of("RevokedEnd_stageInstanceStateId", "id")); + + when(mockRepo.listAllStatesForInstance(workflowInstanceId)) + .thenReturn(List.of(taskReview, policyAgent, grantedAccess, revokedEnd)); + + ResultList result = + resource.listByWorkflowInstanceId(mock(SecurityContext.class), workflowInstanceId, true); + + assertEquals( + 2, result.getData().size(), "only user-task stages should be returned when filtered"); + List names = result.getData().stream().map(s -> s.getStage().getName()).toList(); + assertTrue(names.contains("TaskReview")); + assertTrue(names.contains("GrantedAccess")); + assertEquals(2, result.getPaging().getTotal()); + } + } + + @Test + void filterHandlesStatesWithNullStageOrVariables() { + try (MockedStatic entityMock = mockStatic(Entity.class)) { + WorkflowInstanceStateRepository mockRepo = mock(WorkflowInstanceStateRepository.class); + Authorizer mockAuth = mock(Authorizer.class); + WorkflowInstanceStateResource resource = createResource(entityMock, mockRepo, mockAuth); + + UUID workflowInstanceId = UUID.randomUUID(); + WorkflowInstanceState nullStage = + new WorkflowInstanceState() + .withId(UUID.randomUUID()) + .withWorkflowInstanceId(workflowInstanceId); + WorkflowInstanceState nullVariables = state(workflowInstanceId, "TaskReview", null); + WorkflowInstanceState nullStageName = + new WorkflowInstanceState() + .withId(UUID.randomUUID()) + .withWorkflowInstanceId(workflowInstanceId) + .withStage(new Stage().withVariables(Map.of("TaskReview_transitionId", "approve"))); + + when(mockRepo.listAllStatesForInstance(workflowInstanceId)) + .thenReturn(List.of(nullStage, nullVariables, nullStageName)); + + ResultList result = + resource.listByWorkflowInstanceId(mock(SecurityContext.class), workflowInstanceId, true); + + assertNotNull(result); + assertEquals(0, result.getData().size(), "malformed rows must be filtered out, not crash"); + } + } + + @Test + void endpointReturnsEmptyResultWhenNoStatesExist() { + try (MockedStatic entityMock = mockStatic(Entity.class)) { + WorkflowInstanceStateRepository mockRepo = mock(WorkflowInstanceStateRepository.class); + Authorizer mockAuth = mock(Authorizer.class); + WorkflowInstanceStateResource resource = createResource(entityMock, mockRepo, mockAuth); + + UUID workflowInstanceId = UUID.randomUUID(); + when(mockRepo.listAllStatesForInstance(workflowInstanceId)).thenReturn(List.of()); + + ResultList result = + resource.listByWorkflowInstanceId(mock(SecurityContext.class), workflowInstanceId, false); + + assertNotNull(result); + assertEquals(0, result.getData().size()); + assertEquals(0, result.getPaging().getTotal()); + } + } +}