feat: check modify index before composite structure apply

service/src/main/java/com/netcracker/core/declarative/client/reconciler/BaseCompositeReconciler.java

@@ -1,23 +1,23 @@
 package com.netcracker.core.declarative.client.reconciler;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import com.netcracker.core.declarative.client.rest.Condition;
 import com.netcracker.core.declarative.client.rest.ProcessStatus;
 import com.netcracker.core.declarative.exception.NoopConsulException;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 import com.netcracker.core.declarative.resources.base.CoreCondition;
 import com.netcracker.core.declarative.resources.base.CoreResource;
 import com.netcracker.core.declarative.resources.base.Phase;
 import com.netcracker.core.declarative.resources.composite.Composite;
 import com.netcracker.core.declarative.service.CompositeConsulUpdater;
 import com.netcracker.core.declarative.service.CompositeSpec;
 import com.netcracker.core.declarative.service.CompositeStructureUpdateNotifier;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
-import java.util.Set;
 import java.util.function.Function;
 
 import static com.netcracker.core.declarative.client.constants.Constants.VALIDATED_STEP_NAME;
@@ -77,7 +77,7 @@ public UpdateControl<T> reconcileInternal(T composite) throws Exception {
             }
         }
 
-        Set<String> namespaceList = compositeConsulUpdater.getCompositeMembers(compositeSpec.getCompositeId());
+        CompositeMembersList compositeMembersList = compositeConsulUpdater.getCompositeMembers(compositeSpec.getCompositeId());
 
         log.info("Update XaaSes...");
         for (CompositeStructureUpdateNotifier step : compositeStructureUpdateNotifiers) {
@@ -88,7 +88,7 @@ public UpdateControl<T> reconcileInternal(T composite) throws Exception {
 
             try {
                 log.info("Send composite structure to {}", step.getXaasName());
-                step.notify(compositeSpec.getCompositeId(), namespaceList);
+                step.notify(compositeSpec.getCompositeId(), compositeMembersList.members(), compositeMembersList.modifyIndex());
                 completeStep(composite, stepId);
             } catch (Exception e) {
                 log.error("Notification failed for {}", step.getXaasName(), e);

service/src/main/java/com/netcracker/core/declarative/client/rest/CompositeClient.java

@@ -17,7 +17,9 @@ public interface CompositeClient {
 
     record Request(
             String id,
-            Set<String> namespaces) {
+            Set<String> namespaces,
+            long modifyIndex
+    ) {
     }
 
 }

service/src/main/java/com/netcracker/core/declarative/model/CompositeMembersList.java

@@ -0,0 +1,6 @@
+package com.netcracker.core.declarative.model;
+
+import java.util.Set;
+
+public record CompositeMembersList(long modifyIndex, Set<String> members) {
+}

service/src/main/java/com/netcracker/core/declarative/service/CompositeConsulUpdater.java

@@ -1,12 +1,13 @@
 package com.netcracker.core.declarative.service;
 
-import java.util.Set;
+import com.netcracker.core.declarative.model.CompositeMembersList;
+
 import java.util.concurrent.ExecutionException;
 
 public interface CompositeConsulUpdater {
     void updateCompositeStructureInConsul(CompositeSpec compositeSpec) throws ExecutionException, InterruptedException;
 
     void updateCompositeStructureInConsul(String namespace, CompositeSpec compositeSpec) throws ExecutionException, InterruptedException;
 
-    Set<String> getCompositeMembers(String compositeId) throws ExecutionException, InterruptedException;
+    CompositeMembersList getCompositeMembers(String compositeId) throws ExecutionException, InterruptedException;
 }

service/src/main/java/com/netcracker/core/declarative/service/CompositeConsulUpdaterImpl.java

@@ -1,10 +1,11 @@
 package com.netcracker.core.declarative.service;
 
+import com.netcracker.cloud.consul.provider.common.TokenStorage;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 import io.vertx.ext.consul.*;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import com.netcracker.cloud.consul.provider.common.TokenStorage;
 
 import java.nio.file.Paths;
 import java.util.Collections;
@@ -86,18 +87,21 @@ public void updateCompositeStructureInConsul(String namespace, CompositeSpec com
     }
 
     @Override
-    public Set<String> getCompositeMembers(String compositeId) throws ExecutionException, InterruptedException {
+    public CompositeMembersList getCompositeMembers(String compositeId) throws ExecutionException, InterruptedException {
         String compositeDefinitionRoot = COMPOSITE_STRUCTURE_BASE_PATH_TEMPLATE.formatted(compositeId);
         log.info("Get updated composite structure from consul by path: {}", compositeDefinitionRoot);
         ConsulClient consulClient = consulClientFactory.create(consulTokenStorage.get());
         try {
-            return consulClient.getKeys(compositeDefinitionRoot)
+            KeyValueList keyValueList = consulClient.getValues(compositeDefinitionRoot)
                     .toCompletionStage()
                     .toCompletableFuture()
-                    .get()
+                    .get();
+            Set<String> members = keyValueList
+                    .getList()
                     .stream()
-                    .map(s -> Paths.get(compositeDefinitionRoot).relativize(Paths.get(s)).getParent().toString())
+                    .map(s -> Paths.get(compositeDefinitionRoot).relativize(Paths.get(s.getKey())).getParent().toString())
                     .collect(Collectors.toSet());
+            return new CompositeMembersList(keyValueList.getIndex(), members);
         } finally {
             consulClient.close();
         }

service/src/main/java/com/netcracker/core/declarative/service/CompositeStructureUpdateNotifier.java

@@ -2,11 +2,11 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.netcracker.cloud.core.error.rest.tmf.DefaultTmfErrorResponseConverter;
+import com.netcracker.cloud.core.error.rest.tmf.TmfErrorResponse;
 import com.netcracker.core.declarative.client.rest.CompositeClient;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
-import com.netcracker.cloud.core.error.rest.tmf.DefaultTmfErrorResponseConverter;
-import com.netcracker.cloud.core.error.rest.tmf.TmfErrorResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -23,8 +23,8 @@ public class CompositeStructureUpdateNotifier {
     private final String xaasName;
     private final CompositeClient compositeClient;
 
-    public void notify(String compositeId, Set<String> compositeMembers) {
-        CompositeClient.Request compositeStructure = new CompositeClient.Request(compositeId, compositeMembers);
+    public void notify(String compositeId, Set<String> compositeMembers, long modifyIndex) {
+        CompositeClient.Request compositeStructure = new CompositeClient.Request(compositeId, compositeMembers, modifyIndex);
         log.info("Send request to {} with composite structure {}", xaasName, compositeStructure);
         try (jakarta.ws.rs.core.Response response = compositeClient.structures(compositeStructure)) {
             if (response.getStatusInfo().getStatusCode() == SC_NO_CONTENT) {

service/src/main/java/com/netcracker/core/declarative/service/NoopCompositeConsulUpdaterImpl.java

@@ -1,8 +1,8 @@
 package com.netcracker.core.declarative.service;
 
 import com.netcracker.core.declarative.exception.NoopConsulException;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 
-import java.util.Set;
 import java.util.concurrent.ExecutionException;
 
 public class NoopCompositeConsulUpdaterImpl implements CompositeConsulUpdater {
@@ -18,7 +18,7 @@ public void updateCompositeStructureInConsul(String namespace, CompositeSpec com
     }
 
     @Override
-    public Set<String> getCompositeMembers(String compositeId) {
+    public CompositeMembersList getCompositeMembers(String compositeId) {
         throw new NoopConsulException();
     }
 

service/src/main/resources/application.properties

@@ -14,11 +14,11 @@ quarkus.log.console.format=[%d{yyyy-MM-dd'T'HH:mm:ss.SSS}] [%-5p] [request_id=%X
 quarkus.http.limits.max-header-size=${http.buffer.header.max.size:10240}
 
 quarkus.rest-client.mesh-client-v3.url=http://control-plane:8080
-%dev.quarkus.rest-client.mesh-client-v3.url=http://127.0.0.1:8083
+%dev.quarkus.rest-client.mesh-client-v3.url=http://control-plane:8080
 quarkus.rest-client.maas-client.url=http://maas-agent:8080
-%dev.quarkus.rest-client.maas-client.url=http://127.0.0.1:8084
+%dev.quarkus.rest-client.maas-client.url=http://maas-agent:8080
 quarkus.rest-client.dbaas-client.url=http://dbaas-agent:8080
-%dev.quarkus.rest-client.dbaas-client.url=http://127.0.0.1:8085
+%dev.quarkus.rest-client.dbaas-client.url=http://dbaas-agent:8080
 quarkus.rest-client.idp-extensions-client.url=http://idp-extensions:8080
 %dev.quarkus.rest-client.idp-extensions-client.url=http://127.0.0.1:8087
 quarkus.rest-client.key-manager-client.url=http://key-manager:8080
@@ -35,3 +35,5 @@ cloud.composite.structure.xaas.receivers=${COMPOSITE_STRUCTURE_XAAS_RECEIVERS:db
 cloud.composite.structure.xaas.read-timeout=5000
 cloud.composite.structure.xaas.connect-timeout=1000
 cloud.composite.structure.consul.update-timeout=5000
+
+quarkus.consul-source-config.m2m.enabled=false

service/src/test/java/com/netcracker/core/declarative/client/reconciler/CompositeReconcilerTest.java

@@ -1,6 +1,7 @@
 package com.netcracker.core.declarative.client.reconciler;
 
 import com.netcracker.core.declarative.client.rest.CompositeClient;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 import com.netcracker.core.declarative.resources.base.CoreCondition;
 import com.netcracker.core.declarative.resources.base.CoreResource;
 import com.netcracker.core.declarative.resources.composite.Composite;
@@ -17,6 +18,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static com.netcracker.core.declarative.client.reconciler.CompositeReconciler.DBAAS_NAME;
 import static com.netcracker.core.declarative.client.reconciler.CompositeReconciler.MAAS_NAME;
@@ -26,19 +28,21 @@
 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.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 class CompositeReconcilerTest {
 
     @Test
     void reconcileInternal() throws Exception {
         CompositeClient compositeClient = mock(CompositeClient.class);
         when(compositeClient.structures(any())).thenReturn(Response.noContent().build());
+
+        CompositeConsulUpdater compositeConsulUpdater = mock(CompositeConsulUpdater.class);
+        when(compositeConsulUpdater.getCompositeMembers(any())).thenReturn(new CompositeMembersList(123L, Set.of()));
+
         CompositeReconciler compositeReconciler = new CompositeReconciler(
                 mock(KubernetesClient.class),
-                mock(CompositeConsulUpdater.class),
+                compositeConsulUpdater,
                 List.of(
                         new CompositeStructureUpdateNotifier(MAAS_NAME, compositeClient),
                         new CompositeStructureUpdateNotifier(DBAAS_NAME, compositeClient)
@@ -63,6 +67,31 @@ void reconcileInternal() throws Exception {
     }
 
     @Test
+    void reconcileInternal_modify_index() throws Exception {
+        CompositeClient compositeClient = mock(CompositeClient.class);
+        when(compositeClient.structures(any())).thenReturn(Response.noContent().build());
+
+        CompositeConsulUpdater compositeConsulUpdater = mock(CompositeConsulUpdater.class);
+        when(compositeConsulUpdater.getCompositeMembers(any())).thenReturn(new CompositeMembersList(123L, Set.of("ns-1", "ns-2")));
+
+        CompositeReconciler compositeReconciler = new CompositeReconciler(
+                mock(KubernetesClient.class),
+                compositeConsulUpdater,
+                List.of(
+                        new CompositeStructureUpdateNotifier(MAAS_NAME, compositeClient),
+                        new CompositeStructureUpdateNotifier(DBAAS_NAME, compositeClient)
+                )
+        );
+
+        Composite composite = new Composite();
+        composite.setSpec(new RawExtension(new CompositeSpec("", "ns-1", "", null)));
+        compositeReconciler.reconcileInternal(composite);
+
+        verify(compositeClient, times(2)).structures(new CompositeClient.Request("ns-1", Set.of("ns-1", "ns-2"), 123L));
+
+    }
+
+        @Test
     void reconcileInternal_no_consul() throws Exception {
         KubernetesClient kubernetesClient = mock(KubernetesClient.class);
         NamespaceableResource namespaceableResource = mock(NamespaceableResource.class);
@@ -155,9 +184,12 @@ void reconcileInternal_fail_MaaSUpdate() throws Exception {
         when(namespaceableResource.updateStatus()).thenReturn(mock(CoreResource.class));
         when(kubernetesClient.resource(any(HasMetadata.class))).thenReturn(namespaceableResource);
 
+        CompositeConsulUpdater compositeConsulUpdater = mock(CompositeConsulUpdater.class);
+        when(compositeConsulUpdater.getCompositeMembers(any())).thenReturn(new CompositeMembersList(123L, Set.of()));
+
         CompositeReconciler compositeReconciler = new CompositeReconciler(
                 kubernetesClient,
-                mock(CompositeConsulUpdater.class),
+                compositeConsulUpdater,
                 List.of(new CompositeStructureUpdateNotifier(MAAS_NAME, compositeClient))
         );
 
@@ -192,9 +224,12 @@ void reconcileInternal_MaaSUpdate_fail_response() throws Exception {
         when(namespaceableResource.updateStatus()).thenReturn(mock(CoreResource.class));
         when(kubernetesClient.resource(any(HasMetadata.class))).thenReturn(namespaceableResource);
 
+        CompositeConsulUpdater compositeConsulUpdater = mock(CompositeConsulUpdater.class);
+        when(compositeConsulUpdater.getCompositeMembers(any())).thenReturn(new CompositeMembersList(123L, Set.of()));
+
         CompositeReconciler compositeReconciler = new CompositeReconciler(
                 kubernetesClient,
-                mock(CompositeConsulUpdater.class),
+                compositeConsulUpdater,
                 List.of(new CompositeStructureUpdateNotifier(MAAS_NAME, compositeClient))
         );
 
@@ -232,9 +267,12 @@ void reconcileInternal_MaaSUpdate_fail_tmf_response() throws Exception {
         when(namespaceableResource.updateStatus()).thenReturn(mock(CoreResource.class));
         when(kubernetesClient.resource(any(HasMetadata.class))).thenReturn(namespaceableResource);
 
+        CompositeConsulUpdater compositeConsulUpdater = mock(CompositeConsulUpdater.class);
+        when(compositeConsulUpdater.getCompositeMembers(any())).thenReturn(new CompositeMembersList(123L, Set.of()));
+
         CompositeReconciler compositeReconciler = new CompositeReconciler(
                 kubernetesClient,
-                mock(CompositeConsulUpdater.class),
+                compositeConsulUpdater,
                 List.of(new CompositeStructureUpdateNotifier(MAAS_NAME, compositeClient))
         );
 

service/src/test/java/com/netcracker/core/declarative/service/CompositeConsulUpdaterImplTest.java

@@ -1,6 +1,7 @@
 package com.netcracker.core.declarative.service;
 
 import com.netcracker.core.declarative.client.rest.CompositeClient;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 import io.vertx.core.Future;
 import io.vertx.ext.consul.*;
 import org.junit.jupiter.api.BeforeEach;
@@ -241,19 +242,21 @@ void compositeStructureUpdateStep_BO_BG_SO_BG() throws InterruptedException, Exe
     void getCompositeMembers() throws ExecutionException, InterruptedException {
         String basePath = "composite/first/structure";
 
-        when(consulClient.getKeys(basePath))
-                .thenReturn(Future.succeededFuture(List.of(
-                        basePath + "/first" + "/compositeRole",
-                        basePath + "/second" + "/compositeRole",
-                        basePath + "/third" + "/compositeRole"
-                )));
+        when(consulClient.getValues(basePath))
+                .thenReturn(Future.succeededFuture(new KeyValueList()
+                        .setIndex(123L)
+                        .setList(List.of(
+                                new KeyValue().setKey(basePath + "/first" + "/compositeRole").setValue("val1"),
+                                new KeyValue().setKey(basePath + "/second" + "/compositeRole").setValue("val2"),
+                                new KeyValue().setKey(basePath + "/third" + "/compositeRole").setValue("val3")
+                        ))));
 
         CompositeClient compositeClient = mock(CompositeClient.class);
         when(compositeClient.structures(any())).thenReturn(jakarta.ws.rs.core.Response.noContent().build());
 
         CompositeConsulUpdater compositeConsulUpdater = new CompositeConsulUpdaterImpl("first", consulClientFactory, mock(TokenStorage.class));
-        Set<String> compositeMembers = compositeConsulUpdater.getCompositeMembers("first");
-        assertEquals(Set.of("first", "second", "third"), compositeMembers);
+        CompositeMembersList compositeMembersList = compositeConsulUpdater.getCompositeMembers("first");
+        assertEquals(Set.of("first", "second", "third"), compositeMembersList.members());
     }
 
     private void verifyConsulValueSetTxn(List<TxnOperation> ops, List<String> kvs) {

service/src/test/java/com/netcracker/core/declarative/service/CompositeReconcilerConsulTest.java

@@ -1,13 +1,12 @@
 package com.netcracker.core.declarative.service;
 
 import com.netcracker.core.declarative.ConsulTestResource;
+import com.netcracker.core.declarative.model.CompositeMembersList;
 import io.quarkus.test.common.QuarkusTestResource;
 import io.quarkus.test.junit.QuarkusTest;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
-import java.util.Set;
-
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @QuarkusTest
@@ -20,15 +19,15 @@ public class CompositeReconcilerConsulTest {
     @Test
     public void deleteSubTree() throws Exception {
         compositeConsulUpdater.updateCompositeStructureInConsul("test-namespace-baseline", new CompositeSpec("", "test-namespace-baseline", "", null));
-        Set<String> compositeMembers = compositeConsulUpdater.getCompositeMembers("test-namespace-baseline");
-        assertTrue(compositeMembers.contains("test-namespace-baseline"));
+        CompositeMembersList compositeMembers = compositeConsulUpdater.getCompositeMembers("test-namespace-baseline");
+        assertTrue(compositeMembers.members().contains("test-namespace-baseline"));
 
         compositeConsulUpdater.updateCompositeStructureInConsul("test-namespace", new CompositeSpec("", "test-namespace", "", new CompositeSpec.CompositeSpecBaseline("", "test-namespace-baseline", "")));
         compositeMembers = compositeConsulUpdater.getCompositeMembers("test-namespace-baseline");
-        assertTrue(compositeMembers.contains("test-namespace-baseline"));
+        assertTrue(compositeMembers.members().contains("test-namespace-baseline"));
 
         compositeConsulUpdater.updateCompositeStructureInConsul("test-namespace", new CompositeSpec("", "test-namespace", "", new CompositeSpec.CompositeSpecBaseline("", "test-namespace-baseline", "")));
         compositeMembers = compositeConsulUpdater.getCompositeMembers("test-namespace-baseline");
-        assertTrue(compositeMembers.contains("test-namespace-baseline"));
+        assertTrue(compositeMembers.members().contains("test-namespace-baseline"));
     }
 }