CB-16820 StopStart scaling test instance checker should handle interm…

…ittent syncer status changes

core-api/src/main/java/com/sequenceiq/cloudbreak/api/endpoint/v4/autoscales/AutoscaleV4Endpoint.java

@@ -36,6 +36,7 @@
 import com.sequenceiq.cloudbreak.doc.Notes;
 import com.sequenceiq.cloudbreak.doc.OperationDescriptions.StackOpDescription;
 import com.sequenceiq.cloudbreak.jerseyclient.RetryAndMetrics;
+import com.sequenceiq.flow.api.model.FlowIdentifier;
 
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
@@ -59,14 +60,14 @@ public interface AutoscaleV4Endpoint {
     @Produces(APPLICATION_JSON)
     @ApiOperation(value = StackOpDescription.PUT_START_INSTANCES_BY_ID, produces = APPLICATION_JSON,
             notes = Notes.STACK_NOTES, nickname = "putStackForAutoscaleStartByCrn")
-    void putStackStartInstancesByCrn(@PathParam("crn") String crn, @Valid UpdateStackV4Request updateRequest);
+    FlowIdentifier putStackStartInstancesByCrn(@PathParam("crn") String crn, @Valid UpdateStackV4Request updateRequest);
 
     @PUT
     @Path("/stack/startNodes/name/{name}")
     @Produces(APPLICATION_JSON)
     @ApiOperation(value = StackOpDescription.PUT_START_INSTANCES_BY_ID, produces = APPLICATION_JSON,
             notes = Notes.STACK_NOTES, nickname = "putStackForAutoscaleStartByName")
-    void putStackStartInstancesByName(@PathParam("name") String name, @Valid UpdateStackV4Request updateRequest);
+    FlowIdentifier putStackStartInstancesByName(@PathParam("name") String name, @Valid UpdateStackV4Request updateRequest);
 
     @PUT
     @Path("/stack/crn/{crn}/{userId}/cluster")
@@ -150,7 +151,7 @@ void decommissionInternalInstancesForClusterCrn(@PathParam("crn") String cluster
     @Produces(APPLICATION_JSON)
     @ApiOperation(value = StackOpDescription.STOP_MULTIPLE_INSTANCES_BY_ID_IN_WORKSPACE, produces = APPLICATION_JSON,
             notes = Notes.STACK_NOTES, nickname = "autoscaleStopInstancesByCrn")
-    void stopInstancesForClusterCrn(@PathParam("crn") String clusterCrn,
+    FlowIdentifier stopInstancesForClusterCrn(@PathParam("crn") String clusterCrn,
             @RequestBody @NotEmpty List<String> instanceIds,
             @QueryParam("forced") @DefaultValue("false") Boolean forced,
             @QueryParam("scalingStrategy") ScalingStrategy scalingStrategy);
@@ -160,7 +161,7 @@ void stopInstancesForClusterCrn(@PathParam("crn") String clusterCrn,
     @Produces(APPLICATION_JSON)
     @ApiOperation(value = StackOpDescription.STOP_MULTIPLE_INSTANCES_BY_ID_IN_WORKSPACE, produces = APPLICATION_JSON,
             notes = Notes.STACK_NOTES, nickname = "autoscaleStopInstancesByName")
-    void stopInstancesForClusterName(@PathParam("name") String clusterName,
+    FlowIdentifier stopInstancesForClusterName(@PathParam("name") String clusterName,
             @RequestBody @NotEmpty List<String> instanceIds,
             @QueryParam("forced") @DefaultValue("false") Boolean forced,
             @QueryParam("scalingStrategy") ScalingStrategy scalingStrategy);

core/src/main/java/com/sequenceiq/cloudbreak/controller/v4/AutoscaleV4Controller.java

@@ -53,6 +53,7 @@
 import com.sequenceiq.cloudbreak.service.stack.StackService;
 import com.sequenceiq.cloudbreak.structuredevent.CloudbreakRestRequestThreadLocalService;
 import com.sequenceiq.distrox.v1.distrox.StackOperations;
+import com.sequenceiq.flow.api.model.FlowIdentifier;
 
 @Controller
 @Transactional(TxType.NEVER)
@@ -98,16 +99,16 @@ public void putStack(@TenantAwareParam @ResourceCrn String crn, String userId, @
 
     @Override
     @CheckPermissionByResourceCrn(action = AuthorizationResourceAction.SCALE_DATAHUB)
-    public void putStackStartInstancesByCrn(@TenantAwareParam @ResourceCrn String crn, @Valid UpdateStackV4Request updateRequest) {
-        stackCommonService.putStartInstancesInDefaultWorkspace(NameOrCrn.ofCrn(crn), restRequestThreadLocalService.getRequestedWorkspaceId(), updateRequest,
-                ScalingStrategy.STOPSTART);
+    public FlowIdentifier putStackStartInstancesByCrn(@TenantAwareParam @ResourceCrn String crn, @Valid UpdateStackV4Request updateRequest) {
+        return stackCommonService.putStartInstancesInDefaultWorkspace(NameOrCrn.ofCrn(crn), restRequestThreadLocalService.getRequestedWorkspaceId(),
+                updateRequest, ScalingStrategy.STOPSTART);
     }
 
     @Override
     @CheckPermissionByResourceName(action = AuthorizationResourceAction.SCALE_DATAHUB)
-    public void putStackStartInstancesByName(@ResourceName String name, @Valid UpdateStackV4Request updateRequest) {
-        stackCommonService.putStartInstancesInDefaultWorkspace(NameOrCrn.ofName(name), restRequestThreadLocalService.getRequestedWorkspaceId(), updateRequest,
-                ScalingStrategy.STOPSTART);
+    public FlowIdentifier putStackStartInstancesByName(@ResourceName String name, @Valid UpdateStackV4Request updateRequest) {
+        return stackCommonService.putStartInstancesInDefaultWorkspace(NameOrCrn.ofName(name), restRequestThreadLocalService.getRequestedWorkspaceId(),
+                updateRequest, ScalingStrategy.STOPSTART);
     }
 
     @Override
@@ -156,21 +157,21 @@ public void decommissionInternalInstancesForClusterCrn(@TenantAwareParam @Resour
 
     @Override
     @CheckPermissionByResourceCrn(action = AuthorizationResourceAction.SCALE_DATAHUB)
-    public void stopInstancesForClusterCrn(@TenantAwareParam @ResourceCrn String clusterCrn, @NotEmpty List<String> instanceIds,
+    public FlowIdentifier stopInstancesForClusterCrn(@TenantAwareParam @ResourceCrn String clusterCrn, @NotEmpty List<String> instanceIds,
             Boolean forced, ScalingStrategy scalingStrategy) {
         LOGGER.info("stopInstancesForClusterCrn. ScalingStrategy={}, forced={}, clusterCrn={}, instanceIds=[{}]",
                 scalingStrategy, forced, clusterCrn, instanceIds);
-        stackCommonService.stopMultipleInstancesInWorkspace(NameOrCrn.ofCrn(clusterCrn), restRequestThreadLocalService.getRequestedWorkspaceId(),
+        return stackCommonService.stopMultipleInstancesInWorkspace(NameOrCrn.ofCrn(clusterCrn), restRequestThreadLocalService.getRequestedWorkspaceId(),
                 new HashSet(instanceIds), forced);
     }
 
     @Override
     @CheckPermissionByResourceName(action = AuthorizationResourceAction.SCALE_DATAHUB)
-    public void stopInstancesForClusterName(@ResourceName String clusterName, @NotEmpty List<String> instanceIds,
+    public FlowIdentifier stopInstancesForClusterName(@ResourceName String clusterName, @NotEmpty List<String> instanceIds,
             Boolean forced, ScalingStrategy scalingStrategy) {
         LOGGER.info("stopInstancesForClusterName: ScalingStrategy={}, forced={}, clusterName={}, instanceIds=[{}]",
                 scalingStrategy, forced, clusterName, instanceIds);
-        stackCommonService.stopMultipleInstancesInWorkspace(NameOrCrn.ofName(clusterName), restRequestThreadLocalService.getRequestedWorkspaceId(),
+        return stackCommonService.stopMultipleInstancesInWorkspace(NameOrCrn.ofName(clusterName), restRequestThreadLocalService.getRequestedWorkspaceId(),
                 new HashSet<>(instanceIds), forced);
     }
 

core/src/main/java/com/sequenceiq/cloudbreak/service/StackCommonService.java

@@ -178,7 +178,8 @@ public void putInDefaultWorkspace(String crn, UpdateStackV4Request updateRequest
         put(stack, updateRequest);
     }
 
-    public void putStartInstancesInDefaultWorkspace(NameOrCrn nameOrCrn, Long workspaceId, UpdateStackV4Request updateRequest, ScalingStrategy scalingStrategy) {
+    public FlowIdentifier putStartInstancesInDefaultWorkspace(NameOrCrn nameOrCrn, Long workspaceId, UpdateStackV4Request updateRequest,
+            ScalingStrategy scalingStrategy) {
         LOGGER.info("Received putStack: {}, with scalingStrategy: {}, updateRequest: {}", nameOrCrn, scalingStrategy, updateRequest);
         Optional<Stack> stack = stackService.findStackByNameOrCrnAndWorkspaceId(nameOrCrn, workspaceId);
         if (stack.isEmpty()) {
@@ -188,7 +189,7 @@ public void putStartInstancesInDefaultWorkspace(NameOrCrn nameOrCrn, Long worksp
             throw new BadRequestException("The entitlement for scaling via stop/start is not enabled");
         }
         MDCBuilder.buildMdcContext(stack);
-        putStartInstances(stack.get(), updateRequest, scalingStrategy);
+        return putStartInstances(stack.get(), updateRequest, scalingStrategy);
     }
 
     public FlowIdentifier putStopInWorkspace(NameOrCrn nameOrCrn, Long workspaceId) {

integration-test/src/main/java/com/sequenceiq/it/cloudbreak/action/v1/distrox/DistroXScaleStartInstancesAction.java

@@ -38,9 +38,9 @@ public DistroXTestDto action(TestContext testContext, DistroXTestDto testDto, Cl
         updateStackRequest.setWithClusterEvent(true);
         Log.when(LOGGER, String.format(" Starting instances [%s] for distrox '%s' Compute scaling... ", testDto.getInstanceIdsForAction(), testDto.getName()));
         Log.whenJson(LOGGER, " Distrox Compute scale start instances request: ", testDto.getRequest());
-        client.getDefaultClient()
+        testDto.setFlow("scale start", client.getDefaultClient()
                 .autoscaleEndpoint()
-                .putStackStartInstancesByName(testDto.getName(), updateStackRequest);
+                .putStackStartInstancesByName(testDto.getName(), updateStackRequest));
         StackV4Response stackV4Response = client.getDefaultClient()
                 .distroXV1Endpoint()
                 .getByName(testDto.getName(), new HashSet<>(Arrays.asList("hardware_info", "events")));

integration-test/src/main/java/com/sequenceiq/it/cloudbreak/action/v1/distrox/DistroXScaleStopInstancesAction.java

@@ -22,9 +22,9 @@ public class DistroXScaleStopInstancesAction implements Action<DistroXTestDto, C
     public DistroXTestDto action(TestContext testContext, DistroXTestDto testDto, CloudbreakClient client) throws Exception {
         Log.when(LOGGER, String.format(" Stopping instances [%s] for distrox '%s' Compute scaling... ", testDto.getInstanceIdsForAction(), testDto.getName()));
         Log.whenJson(LOGGER, " Distrox Compute scale stop instances request: ", testDto.getRequest());
-        client.getDefaultClient()
+        testDto.setFlow("scale stop", client.getDefaultClient()
                 .autoscaleEndpoint()
-                .stopInstancesForClusterName(testDto.getName(), testDto.getInstanceIdsForAction(), false, ScalingStrategy.STOPSTART);
+                .stopInstancesForClusterName(testDto.getName(), testDto.getInstanceIdsForAction(), false, ScalingStrategy.STOPSTART));
         StackV4Response stackV4Response = client.getDefaultClient()
                 .distroXV1Endpoint()
                 .getByName(testDto.getName(), new HashSet<>(Arrays.asList("hardware_info", "events")));

integration-test/src/main/java/com/sequenceiq/it/cloudbreak/testcase/e2e/distrox/DistroXResilientScaleTests.java

@@ -63,16 +63,12 @@ public void testPercentageResilientScaleDistrox(TestContext testContext, ITestCo
                 .given(DistroXTestDto.class)
                 .when(distroXTestClient.scale(params.getHostGroup(), params.getScaleUpTarget(), params.getAdjustmentType(), params.getThreshold()))
                 .awaitForFlow()
-                .await(STACK_AVAILABLE)
-                .awaitForHealthyInstances()
                 .then(new DistroxScaleThresholdAssertions(params.getHostGroup(), params.getScaleUpTarget(), params.getThreshold()))
                 .when(distroXTestClient.scale(params.getHostGroup(), params.getScaleDownTarget()))
                 .awaitForFlow()
                 .when(distroXTestClient.scale(params.getHostGroup(), params.getScaleUpTarget(), params.getAdjustmentType(), params.getThreshold()))
                 .then((tc, testDto, client) -> removeInstanceWhileScalingGroup(tc, testDto, client, params))
                 .awaitForFlow()
-                .await(STACK_AVAILABLE)
-                .awaitForHealthyInstances()
                 .when(distroXTestClient.get())
                 .then(new DistroxScaleThresholdAssertions(params.getHostGroup(), params.getScaleUpTarget(), params.getThreshold()))
                 .validate();

integration-test/src/main/java/com/sequenceiq/it/cloudbreak/testcase/e2e/distrox/DistroXStopStartScaleTest.java

@@ -10,7 +10,6 @@
 import org.testng.ITestContext;
 import org.testng.annotations.Test;
 
-import com.sequenceiq.cloudbreak.api.endpoint.v4.stacks.base.InstanceStatus;
 import com.sequenceiq.cloudbreak.common.mappable.CloudPlatform;
 import com.sequenceiq.it.cloudbreak.assertion.distrox.DistroxStopStartScaleDurationAssertions;
 import com.sequenceiq.it.cloudbreak.client.DistroXTestClient;
@@ -72,25 +71,21 @@ public void testStopStartScaleDistroX(TestContext testContext, ITestContext iTes
                     return testDto;
                 })
                 .when(distroXTestClient.scaleStopInstances())
-                .await(STACK_AVAILABLE)
-                .awaitForActionedInstances(InstanceStatus.STOPPED)
+                .awaitForFlow()
                 .then(new DistroxStopStartScaleDurationAssertions(6, false))
                 .when(distroXTestClient.scaleStartInstances(params.getHostGroup(), params.getScaleUpTarget()))
-                .await(STACK_AVAILABLE)
-                .awaitForHealthyInstances()
+                .awaitForFlow()
                 .when(distroXTestClient.get())
                 .then(new DistroxStopStartScaleDurationAssertions(6, true));
         IntStream.range(1, params.getTimes())
                 .forEach(i -> {
                             testContext
                                     .given(DistroXTestDto.class)
                                     .when(distroXTestClient.scaleStopInstances())
-                                    .await(STACK_AVAILABLE)
-                                    .awaitForActionedInstances(InstanceStatus.STOPPED)
+                                    .awaitForFlow()
                                     .then(new DistroxStopStartScaleDurationAssertions(6, false))
                                     .when(distroXTestClient.scaleStartInstances(params.getHostGroup(), params.getScaleUpTarget()))
-                                    .await(STACK_AVAILABLE)
-                                    .awaitForHealthyInstances()
+                                    .awaitForFlow()
                                     .when(distroXTestClient.get())
                                     .then(new DistroxStopStartScaleDurationAssertions(6, true));
                         }