frankgh commented on code in PR #256: URL: https://github.com/apache/cassandra-sidecar/pull/256#discussion_r2364743140
########## integration-framework/src/main/java/org/apache/cassandra/sidecar/lifecycle/InJvmDTestLifecycleProvider.java: ########## @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.TimeUnit; + +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; + +/** + * Manages the lifecycle of JVM Dtest Cassandra instances. + * This should used for integration tests where Cassandra instances are started and stopped Review Comment: minor NIT ```suggestion * This should be used for integration tests where Cassandra instances are started and stopped ``` ########## integration-tests/src/integrationTest/org/apache/cassandra/sidecar/lifecycle/InJvmLifecycleProviderIntegrationTest.java: ########## @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; +import org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase; +import org.apache.cassandra.sidecar.utils.SimpleCassandraVersion; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; + +import static org.assertj.core.api.Assumptions.assumeThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + + +/** + * Tests the {@link InJvmDTestLifecycleProvider} to ensure it can start and stop a Cassandra node + */ +public class InJvmLifecycleProviderIntegrationTest extends SharedClusterSidecarIntegrationTestBase +{ + static final InstanceMetadata LOCALHOST_METADATA = mock(InstanceMetadata.class); + private static final String JVM_LIFECYCLE_TEST_MIN_VERSION = "4.1"; + + @BeforeAll + static void beforeAll() + { + when(LOCALHOST_METADATA.host()).thenReturn("localhost"); + } + Review Comment: can we do the version check _before_ we provision the cluster? this will help speed up tests ``` @Override protected void beforeClusterProvisioning() { // JVM Distributed Test framework contains a bug with restarting nodes in version 4.0 (CASSANDRA-19729) assumeThat(SimpleCassandraVersion.create(testVersion.version())) .withFailMessage("JVM Distributed Test framework contains a bug with restarting nodes in version 4.0 (CASSANDRA-19729)") .isGreaterThanOrEqualTo(SimpleCassandraVersion.create(JVM_LIFECYCLE_TEST_MIN_VERSION)); } ``` ########## server/src/test/java/org/apache/cassandra/sidecar/lifecycle/LifecycleManagerTest.java: ########## @@ -0,0 +1,300 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.vertx.core.Vertx; +import org.apache.cassandra.sidecar.TestResourceReaper; +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; +import org.apache.cassandra.sidecar.common.data.LifecycleCassandraState; +import org.apache.cassandra.sidecar.common.data.LifecycleStatus; +import org.apache.cassandra.sidecar.common.response.LifecycleInfoResponse; +import org.apache.cassandra.sidecar.concurrent.ExecutorPools; +import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl; +import org.apache.cassandra.sidecar.exceptions.LifecycleTaskConflictException; +import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher; +import org.jetbrains.annotations.NotNull; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link LifecycleManager} to validate lifecycle management behavior + */ +class LifecycleManagerTest +{ + private static final String TEST_HOST = "127.0.0.1"; + private static final InstanceMetadata TEST_HOST_META = mock(InstanceMetadata.class); + private static final String TEST_HOST_2 = "127.0.0.2"; + private static final InstanceMetadata TEST_HOST_2_META = mock(InstanceMetadata.class); + + protected Vertx vertx; + protected ExecutorPools executorPools; + protected LifecycleProvider mockLifecycleProvider; + protected InstanceMetadataFetcher metadataFetcher = mock(InstanceMetadataFetcher.class); + + @BeforeEach + void setup() + { + vertx = Vertx.vertx(); + executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); + mockLifecycleProvider = mock(LifecycleProvider.class); + when(metadataFetcher.instance(TEST_HOST)).thenReturn(TEST_HOST_META); + when(metadataFetcher.instance(TEST_HOST_2)).thenReturn(TEST_HOST_2_META); + } + + @AfterEach + void cleanup() + { + TestResourceReaper.create().with(vertx).with(executorPools).close(); + } + + @Test + void testGetLifecycleInfoWithNoTaskSubmitted() + { + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + + LifecycleInfoResponse response = lifecycleManager.getLifecycleInfo(TEST_HOST); + assertThat(response.currentState()).isEqualTo(LifecycleCassandraState.STOPPED); + assertThat(response.desiredState()).isEqualTo(LifecycleCassandraState.UNKNOWN); + assertThat(response.status()).isEqualTo(LifecycleStatus.UNDEFINED); + assertThat(response.lastUpdate()).isEqualTo("No lifecycle task submitted for this instance yet."); + } + + @Test + void testSubmittedTaskSucceeds() throws LifecycleTaskConflictException, InterruptedException + { + // Submit slow start task + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + CountDownLatch startLatch = slowCassandraStart(); + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + + LifecycleInfoResponse actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + LifecycleInfoResponse expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.STOPPED, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGING, + "Submitting start task for instance"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Wait for the task to complete + startLatch.countDown(); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(true); + Thread.sleep(200); + + // Check status was updated + actualResponse = lifecycleManager.getLifecycleInfo(TEST_HOST); + expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.RUNNING, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGED, + "Instance has started"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Attempt to start the instance again, should be no-op since instance is already running + actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + assertThat(actualResponse).isEqualTo(expectedResponse); + } + + @Test + void testSubmittedTaskFails() throws LifecycleTaskConflictException, InterruptedException + { + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + + String errorMessage = "Cannot find Cassandra executable to start instance."; + doThrow(new RuntimeException(errorMessage)).when(mockLifecycleProvider).start(TEST_HOST_META); + + lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + Thread.sleep(200); Review Comment: NIT about the usage of thread.sleep ########## server/src/test/java/org/apache/cassandra/sidecar/lifecycle/LifecycleManagerTest.java: ########## @@ -0,0 +1,300 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.vertx.core.Vertx; +import org.apache.cassandra.sidecar.TestResourceReaper; +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; +import org.apache.cassandra.sidecar.common.data.LifecycleCassandraState; +import org.apache.cassandra.sidecar.common.data.LifecycleStatus; +import org.apache.cassandra.sidecar.common.response.LifecycleInfoResponse; +import org.apache.cassandra.sidecar.concurrent.ExecutorPools; +import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl; +import org.apache.cassandra.sidecar.exceptions.LifecycleTaskConflictException; +import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher; +import org.jetbrains.annotations.NotNull; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link LifecycleManager} to validate lifecycle management behavior + */ +class LifecycleManagerTest +{ + private static final String TEST_HOST = "127.0.0.1"; + private static final InstanceMetadata TEST_HOST_META = mock(InstanceMetadata.class); + private static final String TEST_HOST_2 = "127.0.0.2"; + private static final InstanceMetadata TEST_HOST_2_META = mock(InstanceMetadata.class); + + protected Vertx vertx; + protected ExecutorPools executorPools; + protected LifecycleProvider mockLifecycleProvider; + protected InstanceMetadataFetcher metadataFetcher = mock(InstanceMetadataFetcher.class); + + @BeforeEach + void setup() + { + vertx = Vertx.vertx(); + executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); + mockLifecycleProvider = mock(LifecycleProvider.class); + when(metadataFetcher.instance(TEST_HOST)).thenReturn(TEST_HOST_META); + when(metadataFetcher.instance(TEST_HOST_2)).thenReturn(TEST_HOST_2_META); + } + + @AfterEach + void cleanup() + { + TestResourceReaper.create().with(vertx).with(executorPools).close(); + } + + @Test + void testGetLifecycleInfoWithNoTaskSubmitted() + { + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + + LifecycleInfoResponse response = lifecycleManager.getLifecycleInfo(TEST_HOST); + assertThat(response.currentState()).isEqualTo(LifecycleCassandraState.STOPPED); + assertThat(response.desiredState()).isEqualTo(LifecycleCassandraState.UNKNOWN); + assertThat(response.status()).isEqualTo(LifecycleStatus.UNDEFINED); + assertThat(response.lastUpdate()).isEqualTo("No lifecycle task submitted for this instance yet."); + } + + @Test + void testSubmittedTaskSucceeds() throws LifecycleTaskConflictException, InterruptedException + { + // Submit slow start task + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + CountDownLatch startLatch = slowCassandraStart(); + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + + LifecycleInfoResponse actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + LifecycleInfoResponse expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.STOPPED, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGING, + "Submitting start task for instance"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Wait for the task to complete + startLatch.countDown(); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(true); + Thread.sleep(200); + + // Check status was updated + actualResponse = lifecycleManager.getLifecycleInfo(TEST_HOST); + expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.RUNNING, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGED, + "Instance has started"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Attempt to start the instance again, should be no-op since instance is already running + actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + assertThat(actualResponse).isEqualTo(expectedResponse); + } + + @Test + void testSubmittedTaskFails() throws LifecycleTaskConflictException, InterruptedException + { + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + + String errorMessage = "Cannot find Cassandra executable to start instance."; + doThrow(new RuntimeException(errorMessage)).when(mockLifecycleProvider).start(TEST_HOST_META); + + lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + Thread.sleep(200); + + LifecycleInfoResponse response = lifecycleManager.getLifecycleInfo(TEST_HOST); + assertThat(response.currentState()).isEqualTo(LifecycleCassandraState.STOPPED); + assertThat(response.desiredState()).isEqualTo(LifecycleCassandraState.RUNNING); + assertThat(response.status()).isEqualTo(LifecycleStatus.DIVERGED); + assertThat(response.lastUpdate()).isEqualTo(String.format("Failed to start instance 127.0.0.1: %s", errorMessage)); + + lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + reset(mockLifecycleProvider); + + Thread.sleep(200); Review Comment: NIT about the usage of thread.sleep ########## integration-tests/src/integrationTest/org/apache/cassandra/sidecar/lifecycle/LifecycleProviderIntegrationTester.java: ########## @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.google.common.util.concurrent.Uninterruptibles; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.vertx.core.buffer.Buffer; +import io.vertx.core.json.JsonObject; +import io.vertx.ext.web.client.HttpResponse; +import io.vertx.ext.web.client.WebClient; +import org.apache.cassandra.sidecar.testing.SharedClusterIntegrationTestBase; + +import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Utility class to test different lifecycle provider implementations. + */ +public class LifecycleProviderIntegrationTester +{ + protected final Logger logger = LoggerFactory.getLogger(SharedClusterIntegrationTestBase.class); Review Comment: Do we want the `SharedClusterIntegrationTestBase` logger here? ```suggestion protected final Logger logger = LoggerFactory.getLogger(LifecycleProviderIntegrationTester.class); ``` ########## server/src/test/java/org/apache/cassandra/sidecar/lifecycle/LifecycleManagerTest.java: ########## @@ -0,0 +1,300 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.vertx.core.Vertx; +import org.apache.cassandra.sidecar.TestResourceReaper; +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; +import org.apache.cassandra.sidecar.common.data.LifecycleCassandraState; +import org.apache.cassandra.sidecar.common.data.LifecycleStatus; +import org.apache.cassandra.sidecar.common.response.LifecycleInfoResponse; +import org.apache.cassandra.sidecar.concurrent.ExecutorPools; +import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl; +import org.apache.cassandra.sidecar.exceptions.LifecycleTaskConflictException; +import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher; +import org.jetbrains.annotations.NotNull; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link LifecycleManager} to validate lifecycle management behavior + */ +class LifecycleManagerTest +{ + private static final String TEST_HOST = "127.0.0.1"; + private static final InstanceMetadata TEST_HOST_META = mock(InstanceMetadata.class); + private static final String TEST_HOST_2 = "127.0.0.2"; + private static final InstanceMetadata TEST_HOST_2_META = mock(InstanceMetadata.class); + + protected Vertx vertx; + protected ExecutorPools executorPools; + protected LifecycleProvider mockLifecycleProvider; + protected InstanceMetadataFetcher metadataFetcher = mock(InstanceMetadataFetcher.class); + + @BeforeEach + void setup() + { + vertx = Vertx.vertx(); + executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); + mockLifecycleProvider = mock(LifecycleProvider.class); + when(metadataFetcher.instance(TEST_HOST)).thenReturn(TEST_HOST_META); + when(metadataFetcher.instance(TEST_HOST_2)).thenReturn(TEST_HOST_2_META); + } + + @AfterEach + void cleanup() + { + TestResourceReaper.create().with(vertx).with(executorPools).close(); + } + + @Test + void testGetLifecycleInfoWithNoTaskSubmitted() + { + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + + LifecycleInfoResponse response = lifecycleManager.getLifecycleInfo(TEST_HOST); + assertThat(response.currentState()).isEqualTo(LifecycleCassandraState.STOPPED); + assertThat(response.desiredState()).isEqualTo(LifecycleCassandraState.UNKNOWN); + assertThat(response.status()).isEqualTo(LifecycleStatus.UNDEFINED); + assertThat(response.lastUpdate()).isEqualTo("No lifecycle task submitted for this instance yet."); + } + + @Test + void testSubmittedTaskSucceeds() throws LifecycleTaskConflictException, InterruptedException + { + // Submit slow start task + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(false); + CountDownLatch startLatch = slowCassandraStart(); + LifecycleManager lifecycleManager = new LifecycleManager(metadataFetcher, mockLifecycleProvider, executorPools); + + LifecycleInfoResponse actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + LifecycleInfoResponse expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.STOPPED, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGING, + "Submitting start task for instance"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Wait for the task to complete + startLatch.countDown(); + when(mockLifecycleProvider.isRunning(TEST_HOST_META)).thenReturn(true); + Thread.sleep(200); + + // Check status was updated + actualResponse = lifecycleManager.getLifecycleInfo(TEST_HOST); + expectedResponse = new LifecycleInfoResponse(LifecycleCassandraState.RUNNING, LifecycleCassandraState.RUNNING, + LifecycleStatus.CONVERGED, + "Instance has started"); + assertThat(actualResponse).isEqualTo(expectedResponse); + + // Attempt to start the instance again, should be no-op since instance is already running + actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); + assertThat(actualResponse).isEqualTo(expectedResponse); Review Comment: We like to avoid doing thread sleep and we actually have a helper method that allows us to wait in a loop ```suggestion // Check status was updated LifecycleInfoResponse actualResponseAfterStart = lifecycleManager.getLifecycleInfo(TEST_HOST); LifecycleInfoResponse expectedResponseAfterStart = new LifecycleInfoResponse(LifecycleCassandraState.RUNNING, LifecycleCassandraState.RUNNING, LifecycleStatus.CONVERGED, "Instance has started"); loopAssert(1, 10, () -> assertThat(actualResponseAfterStart).isEqualTo(expectedResponseAfterStart)); // Attempt to start the instance again, should be no-op since instance is already running actualResponse = lifecycleManager.updateDesiredState(TEST_HOST, LifecycleCassandraState.RUNNING); assertThat(actualResponse).isEqualTo(expectedResponseAfterStart); ``` ########## integration-tests/src/integrationTest/org/apache/cassandra/sidecar/lifecycle/LifecycleProviderIntegrationTester.java: ########## @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.lifecycle; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.google.common.util.concurrent.Uninterruptibles; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.vertx.core.buffer.Buffer; +import io.vertx.core.json.JsonObject; +import io.vertx.ext.web.client.HttpResponse; +import io.vertx.ext.web.client.WebClient; +import org.apache.cassandra.sidecar.testing.SharedClusterIntegrationTestBase; + +import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Utility class to test different lifecycle provider implementations. + */ +public class LifecycleProviderIntegrationTester +{ + protected final Logger logger = LoggerFactory.getLogger(SharedClusterIntegrationTestBase.class); + + static final int TIMEOUT_SECONDS = 120; + + final WebClient client; + final String sidecarHost; + final int sidecarPort; + final Runnable cassandraNodeCrasher; + + public LifecycleProviderIntegrationTester(WebClient client, String sidecarHost, int sidecarPort, Runnable cassandraNodeCrasher) + { + this.client = client; + this.sidecarHost = sidecarHost; + this.sidecarPort = sidecarPort; + this.cassandraNodeCrasher = cassandraNodeCrasher; + } + + void testLifecycleProviderStartAndStopAndRecoveryAfterCrash() throws Exception + { + // Check CQL health is NOT_OK, since node is not started + assertThat(getCqlStatus()).isEqualTo("NOT_OK"); + + // GET /lifecycle when node has not started and no intent has been submitted + HttpResponse<Buffer> lifecycleInfoOnStartup = getLifecycle(); + assertExpectedLifecycleResponse(lifecycleInfoOnStartup, OK.code(), + "STOPPED", "UNKNOWN", "UNDEFINED", + "No lifecycle task submitted for this instance yet."); + + // Start node via PUT /lifecycle and wait for node to be up + HttpResponse<Buffer> lifecycleUpdateStart = putLifecycle("start"); + assertExpectedLifecycleResponse(lifecycleUpdateStart, ACCEPTED.code(), + "STOPPED", "RUNNING", "CONVERGING", "Submitting start task for instance"); + + waitForLastUpdateToConverge("Instance has started", TIMEOUT_SECONDS); + waitForCqlStatus("OK", TIMEOUT_SECONDS); + + // GET /lifecycle after node has started up + HttpResponse<Buffer> lifecycleInfoAfterStartup = getLifecycle(); + assertExpectedLifecycleResponse(lifecycleInfoAfterStartup, OK.code(), + "RUNNING", "RUNNING", "CONVERGED", "Instance has started"); + + // Try starting node with PUT /lifecycle again + HttpResponse<Buffer> lifecycleInfoAfterStartAgain = putLifecycle("start"); + assertExpectedLifecycleResponse(lifecycleInfoAfterStartAgain, OK.code(), + "RUNNING", "RUNNING", "CONVERGED", "Instance has started"); + + // Simulate node crashing by directly calling the lifecycle provider's stop method + cassandraNodeCrasher.run(); + waitForLastUpdateToConverge(String.format("Instance %s has unexpectedly diverged from the desired state RUNNING to STOPPED.", sidecarHost), TIMEOUT_SECONDS); + + // CQL status should be down since instance is crashed + waitForCqlStatus("NOT_OK", TIMEOUT_SECONDS); + + // GET /lifecycle after node has crashed + HttpResponse<Buffer> lifecycleInfoAfterCrash = getLifecycle(); + assertExpectedLifecycleResponse(lifecycleInfoAfterCrash, OK.code(), + "STOPPED", "RUNNING", "DIVERGED", + String.format("Instance %s has unexpectedly diverged from the desired state RUNNING to STOPPED.", sidecarHost)); + + // Try starting node with PUT /lifecycle again + HttpResponse<Buffer> lifecycleInfoOnStartAfterCrash = putLifecycle("start"); + assertExpectedLifecycleResponse(lifecycleInfoOnStartAfterCrash, ACCEPTED.code(), + "STOPPED", "RUNNING", "CONVERGING", "Submitting start task for instance"); + waitForLastUpdateToConverge("Instance has started", TIMEOUT_SECONDS); + waitForCqlStatus("OK", TIMEOUT_SECONDS); + + // Confirm that the node is up again with GET /lifecycle + HttpResponse<Buffer> lifecycleInfoAfterRecovery = getLifecycle(); + assertExpectedLifecycleResponse(lifecycleInfoAfterRecovery, OK.code(), + "RUNNING", "RUNNING", "CONVERGED", "Instance has started"); + + // Stop node via PUT /lifecycle + HttpResponse<Buffer> lifecycleUpdateStop = putLifecycle("stop"); + assertExpectedLifecycleResponse(lifecycleUpdateStop, ACCEPTED.code(), + "RUNNING", "STOPPED", "CONVERGING", "Submitting stop task for instance"); + + // Allow time for the async stop task to complete + waitForLastUpdateToConverge("Instance has stopped", TIMEOUT_SECONDS); + + // GET /lifecycle after node has stopped + HttpResponse<Buffer> lifecycleInfoAfterStop = getLifecycle(); + assertExpectedLifecycleResponse(lifecycleInfoAfterStop, OK.code(), + "STOPPED", "STOPPED", "CONVERGED", "Instance has stopped"); + + // CQL status should be down since instance is stopped + assertThat(getCqlStatus()).isEqualTo("NOT_OK"); + + // Try stopping node with PUT /lifecycle again, should be idempotent + HttpResponse<Buffer> lifecycleInfoOnStopAgain = putLifecycle("stop"); + assertExpectedLifecycleResponse(lifecycleInfoOnStopAgain, OK.code(), + "STOPPED", "STOPPED", "CONVERGED", "Instance has stopped"); + } + + private void assertExpectedLifecycleResponse( + HttpResponse<Buffer> response, + int expectedStatusCode, + String currentState, + String desiredState, + String status, + String lastUpdate) + { + assertThat(response.statusCode()).isEqualTo(expectedStatusCode); + JsonObject body = response.bodyAsJsonObject(); + assertThat(body.getString("current_state")).isEqualTo(currentState); + assertThat(body.getString("desired_state")).isEqualTo(desiredState); + assertThat(body.getString("status")).isEqualTo(status); + assertThat(body.getString("last_update")).isEqualTo(lastUpdate); + } + + private HttpResponse<Buffer> getLifecycle() + { + return getBlocking(client + .get(sidecarPort, sidecarHost, "/api/v1/cassandra/lifecycle") + .send()); + } + + private HttpResponse<Buffer> putLifecycle(String desiredState) + { + return getBlocking(client + .put(sidecarPort, sidecarHost, "/api/v1/cassandra/lifecycle") + .sendBuffer(Buffer.buffer("{\"state\":\"" + desiredState + "\"}"))); Review Comment: NIT: this is a little bit easier to read and a bit simpler to build using `io.vertx.core.json.JsonObject` ```suggestion .sendBuffer(JsonObject.of("state", desiredState).toBuffer())); ``` ########## server/src/test/java/org/apache/cassandra/sidecar/handlers/LifecycleUpdateHandlerTest.java: ########## @@ -0,0 +1,205 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.sidecar.handlers; + +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import com.google.inject.AbstractModule; +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; +import com.google.inject.Provides; +import com.google.inject.Singleton; +import com.google.inject.util.Modules; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.vertx.core.Vertx; +import io.vertx.ext.web.client.WebClient; +import io.vertx.junit5.VertxExtension; +import io.vertx.junit5.VertxTestContext; +import org.apache.cassandra.sidecar.TestModule; +import org.apache.cassandra.sidecar.TestResourceReaper; +import org.apache.cassandra.sidecar.common.data.LifecycleCassandraState; +import org.apache.cassandra.sidecar.common.data.LifecycleStatus; +import org.apache.cassandra.sidecar.common.response.LifecycleInfoResponse; +import org.apache.cassandra.sidecar.exceptions.LifecycleTaskConflictException; +import org.apache.cassandra.sidecar.lifecycle.LifecycleManager; +import org.apache.cassandra.sidecar.modules.SidecarModules; +import org.apache.cassandra.sidecar.server.Server; + +import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED; +import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST; +import static io.netty.handler.codec.http.HttpResponseStatus.CONFLICT; +import static io.vertx.core.buffer.Buffer.buffer; +import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Test for {@link LifecycleUpdateHandler} + */ +@ExtendWith(VertxExtension.class) +public class LifecycleUpdateHandlerTest +{ + Vertx vertx; + Server server; + LifecycleManager mockLifecycleManager = mock(LifecycleManager.class); + + @BeforeEach + void before() throws InterruptedException + { + Injector injector; + Module testOverride = Modules.override(new TestModule()).with(new LifecycleUpdateHandlerTestModule()); + injector = Guice.createInjector(Modules.override(SidecarModules.all()).with(testOverride)); + vertx = injector.getInstance(Vertx.class); + server = injector.getInstance(Server.class); + VertxTestContext context = new VertxTestContext(); + server.start().onSuccess(s -> context.completeNow()).onFailure(context::failNow); + context.awaitCompletion(5, TimeUnit.SECONDS); + reset(mockLifecycleManager); + } + + @AfterEach + void after() throws InterruptedException + { + getBlocking(TestResourceReaper.create().with(server).close(), 60, TimeUnit.SECONDS, "Closing server"); + } + + @Test + void testSuccessfulPutWithAcceptedResponse(VertxTestContext ctx) throws LifecycleTaskConflictException + { + WebClient client = WebClient.create(vertx); + String payload = "{\"state\":\"start\"}"; Review Comment: alternatively can use JsonObject -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
