Add integration test for backup and restore

[deepthidevaki]

Test following scenarios:

• Cluster multiple partitions
  • Backup can be taken, get status returns correct status and backup can be restored
  • Backup is also triggered via inter partition messages.
• Cluster with multiple partitions and replication factor 3
  • After a leader change, get status of old backup returns correct status
  • After a leader change, new backup can be taken
  • After a leader change, a backup that is started by the previous broker but not completed is marked as failed

[npepinpe]

Here's an acceptance I wrote for the take backup API - as it may be the wrong level for testing the endpoint, I'll leave it here as a template of a QA test using containers, a real backup store, and the actuators:

qa/integration-tests/src/test/java/io/camunda/zeebe/it/backup/BackupApiIT.java

/*
 * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
 * one or more contributor license agreements. See the NOTICE file distributed
 * with this work for additional information regarding copyright ownership.
 * Licensed under the Zeebe Community License 1.1. You may not use this file
 * except in compliance with the Zeebe Community License 1.1.
 */
package io.camunda.zeebe.it.backup;

import static org.assertj.core.api.Assertions.assertThat;

import io.camunda.zeebe.backup.api.BackupStatus;
import io.camunda.zeebe.backup.api.BackupStatusCode;
import io.camunda.zeebe.backup.common.BackupIdentifierImpl;
import io.camunda.zeebe.backup.s3.S3BackupConfig;
import io.camunda.zeebe.backup.s3.S3BackupStore;
import io.camunda.zeebe.qa.util.actuator.BackupActuator;
import io.camunda.zeebe.qa.util.actuator.BackupActuator.TakeBackupResponse;
import io.camunda.zeebe.qa.util.testcontainers.ContainerLogsDumper;
import io.camunda.zeebe.qa.util.testcontainers.ZeebeTestContainerDefaults;
import io.zeebe.containers.ZeebeBrokerNode;
import io.zeebe.containers.ZeebeNode;
import io.zeebe.containers.ZeebePort;
import io.zeebe.containers.cluster.ZeebeCluster;
import io.zeebe.containers.engine.ContainerEngine;
import java.time.Duration;
import java.util.Objects;
import org.agrona.CloseHelper;
import org.apache.commons.lang3.RandomStringUtils;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.containers.localstack.LocalStackContainer.Service;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

/**
 * Acceptance tests for the backup management API. Tests here should interact with the backups
 * primarily via the management API, and occasionally assert results on the configured backup store.
 *
 * <p>The tests run against a cluster of 2 brokers and 1 gateway, no embedded gateways, two
 * partitions and replication factor of 1. This allows us to test that requests are correctly fanned
 * out across the gateway, since each broker is guaranteed to be leader of a partition.
 *
 * <p>NOTE: this does not test the consistency of backups, nor that partition leaders correctly
 * maintain consistency via checkpoint records. Other test suites should be set up for this.
 */
@Testcontainers
final class BackupApiIT {
  private static final int S3_PORT = 4566;
  private static final Network NETWORK = Network.newNetwork();

  @Container
  private static final LocalStackContainer S3 =
      new LocalStackContainer(DockerImageName.parse("localstack/localstack:0.14.5"))
          .withNetwork(NETWORK)
          .withServices(Service.S3);

  private final S3BackupConfig config =
      S3BackupConfig.from(
          RandomStringUtils.randomAlphabetic(10).toLowerCase(),
          S3.getEndpointOverride(Service.S3).toString(),
          S3.getRegion(),
          S3.getAccessKey(),
          S3.getSecretKey());
  private final S3BackupStore store = new S3BackupStore(config);

  private final ZeebeCluster cluster =
      ZeebeCluster.builder()
          .withImage(ZeebeTestContainerDefaults.defaultTestImage())
          .withNetwork(NETWORK)
          .withBrokersCount(2)
          .withGatewaysCount(1)
          .withReplicationFactor(1)
          .withPartitionsCount(2)
          .withEmbeddedGateway(false)
          .withBrokerConfig(this::configureBroker)
          .withNodeConfig(this::configureNode)
          .build();

  @RegisterExtension
  @SuppressWarnings("unused")
  final ContainerLogsDumper logsWatcher = new ContainerLogsDumper(cluster::getNodes);

  @Container
  private final ContainerEngine engine =
      ContainerEngine.builder().withAutoAcknowledge(true).withCluster(cluster).build();

  @AfterAll
  static void afterAll() {
    CloseHelper.quietCloseAll(NETWORK);
  }

  @BeforeEach
  void beforeEach() {
    try (final var s3Client = S3BackupStore.buildClient(config)) {
      s3Client.createBucket(builder -> builder.bucket(config.bucketName()).build()).join();
    }
  }

  @Test
  void shouldTakeBackup() {
    // given
    final var actuator = BackupActuator.of(cluster.getAvailableGateway());
    try (final var client = engine.createClient()) {
      client.newPublishMessageCommand().messageName("name").correlationKey("key").send().join();
    }

    // when
    final var response = actuator.take(1L);

    // then
    assertThat(response).isEqualTo(new TakeBackupResponse(1L));
    Awaitility.await("until a backup exists with the given ID")
        .atMost(Duration.ofSeconds(30))
        .untilAsserted(this::assertBackupCompleteOnAllPartitions);
  }

  private void assertBackupCompleteOnAllPartitions() {
    // TODO: this will be replaced by the status API later
    for (int partitionId = 1; partitionId < 2; partitionId++) {
      assertBackupCompleteForPartition(partitionId);
    }
  }

  private void assertBackupCompleteForPartition(final int partitionId) {
    final var backupId = new BackupIdentifierImpl(0, partitionId, 1);
    final var status = store.getStatus(backupId);

    assertThat(status)
        .succeedsWithin(Duration.ofSeconds(30))
        .extracting(BackupStatus::id, BackupStatus::statusCode)
        .containsExactly(backupId, BackupStatusCode.COMPLETED);
  }

  private void configureBroker(final ZeebeBrokerNode<?> broker) {
    broker
        .withEnv("ZEEBE_BROKER_EXPERIMENTAL_FEATURES_ENABLEBACKUP", "true")
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_STORE", "S3")
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_S3_BUCKETNAME", config.bucketName())
        // force using path-style instead of host-prefix, as otherwise the bucket will not be
        // resolved
        .withEnv(
            "ZEEBE_BROKER_DATA_BACKUP_S3_ENDPOINT",
            "http://%s:%d".formatted(getIpAddress(), S3_PORT))
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_S3_REGION", S3.getRegion())
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_S3_ACCESSKEY", S3.getAccessKey())
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_S3_SECRETKEY", S3.getSecretKey());
  }

  private void configureNode(final ZeebeNode<?> node) {
    node.withEnv("MANAGEMENT_ENDPOINTS_WEB_EXPOSURE_INCLUDE", "*")
        .withEnv("MANAGEMENT_ENDPOINTS_BACKUPS_ENABLED", "true");
    node.addExposedPort(ZeebePort.MONITORING.getPort());
  }

  private String getIpAddress() {
    return S3.getCurrentContainerInfo().getNetworkSettings().getNetworks().values().stream()
        .filter(n -> Objects.equals(n.getNetworkID(), NETWORK.getId()))
        .findFirst()
        .orElseThrow()
        .getIpAddress();
  }
}

I also had a variant which uses the log as the source of truth, but I think the approach of checking the store is better for acceptance tests (though long term it probably makes more sense for these tests to use the take/restore and assert the state at the end).

qa/integration-tests/src/test/java/io/camunda/zeebe/it/management/BackupEndpointIT.java

/*
 * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
 * one or more contributor license agreements. See the NOTICE file distributed
 * with this work for additional information regarding copyright ownership.
 * Licensed under the Zeebe Community License 1.1. You may not use this file
 * except in compliance with the Zeebe Community License 1.1.
 */
package io.camunda.zeebe.it.management;

import static org.assertj.core.api.Assertions.assertThat;

import io.camunda.zeebe.process.test.assertions.BpmnAssert;
import io.camunda.zeebe.protocol.record.Record;
import io.camunda.zeebe.protocol.record.ValueType;
import io.camunda.zeebe.protocol.record.intent.management.CheckpointIntent;
import io.camunda.zeebe.protocol.record.value.management.CheckpointRecordValue;
import io.camunda.zeebe.qa.util.actuator.BackupActuator;
import io.camunda.zeebe.qa.util.actuator.BackupActuator.TakeBackupResponse;
import io.camunda.zeebe.qa.util.testcontainers.ContainerLogsDumper;
import io.camunda.zeebe.qa.util.testcontainers.ZeebeTestContainerDefaults;
import io.zeebe.containers.ZeebeBrokerNode;
import io.zeebe.containers.ZeebeNode;
import io.zeebe.containers.ZeebePort;
import io.zeebe.containers.cluster.ZeebeCluster;
import io.zeebe.containers.engine.ContainerEngine;
import java.time.Duration;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.agrona.CloseHelper;
import org.assertj.core.groups.Tuple;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.testcontainers.containers.Network;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

/**
 * Integration tests for the backup management endpoint API. Tests here should interact strictly
 * with the management endpoint and the exported log. The tests run without a backup store, as the
 * goal isn't to verify that that part works, but just that the communication between the endpoint
 * and the cluster works as expected.
 *
 * <p>The tests run against a cluster of 2 brokers and 1 gateway, no embedded gateways, two
 * partitions and replication factor of 1. This allows us to test that requests are correctly fanned
 * out across the gateway, since each broker is guaranteed to be leader of a partition.
 */
@Testcontainers
final class BackupEndpointIT {
  private static final Network NETWORK = Network.newNetwork();

  private final ZeebeCluster cluster =
      ZeebeCluster.builder()
          .withImage(ZeebeTestContainerDefaults.defaultTestImage())
          .withNetwork(NETWORK)
          .withBrokersCount(2)
          .withGatewaysCount(1)
          .withReplicationFactor(1)
          .withPartitionsCount(2)
          .withEmbeddedGateway(false)
          .withNodeConfig(this::configureNode)
          .withBrokerConfig(this::configureBroker)
          .build();

  @RegisterExtension
  @SuppressWarnings("unused")
  final ContainerLogsDumper logsWatcher = new ContainerLogsDumper(cluster::getNodes);

  @Container
  private final ContainerEngine engine =
      ContainerEngine.builder()
          .withAutoAcknowledge(true)
          .withCluster(cluster)
          .withGracePeriod(Duration.ofSeconds(10))
          .build();

  @AfterAll
  static void afterAll() {
    CloseHelper.quietCloseAll(NETWORK);
  }

  @Test
  void shouldTakeBackup() {
    // given
    final var actuator = BackupActuator.of(cluster.getAvailableGateway());
    try (final var client = engine.createClient()) {
      client.newPublishMessageCommand().messageName("name").correlationKey("key").send().join();
    }

    // when
    final var response = actuator.take(1L);

    // then
    assertThat(response).isEqualTo(new TakeBackupResponse(1L));
    final var checkpointRecords =
        getExportedRecords().filter(r -> r.getIntent() == CheckpointIntent.CREATED).toList();
    assertThat(checkpointRecords)
        .hasSize(2)
        .extracting(Record::getPartitionId, r -> r.getValue().getCheckpointId())
        .containsExactly(Tuple.tuple(1, 1L), Tuple.tuple(2, 1L));
  }

  private void configureBroker(final ZeebeBrokerNode<?> broker) {
    broker
        .withEnv("ZEEBE_BROKER_EXPERIMENTAL_FEATURES_ENABLEBACKUP", "true")
        .withEnv("ZEEBE_BROKER_DATA_BACKUP_STORE", "NONE");
  }

  private void configureNode(final ZeebeNode<?> node) {
    node.withEnv("MANAGEMENT_ENDPOINTS_WEB_EXPOSURE_INCLUDE", "*")
        .withEnv("MANAGEMENT_ENDPOINTS_BACKUPS_ENABLED", "true");
    node.addExposedPort(ZeebePort.MONITORING.getPort());
  }

  @SuppressWarnings("unchecked")
  private Stream<Record<CheckpointRecordValue>> getExportedRecords() {
    final var exportedRecords = BpmnAssert.getRecordStream().records();
    return StreamSupport.stream(exportedRecords.spliterator(), false)
        .filter(r -> r.getValueType() == ValueType.CHECKPOINT)
        .map(r -> (Record<CheckpointRecordValue>) r);
  }
}