OAK-12215 Add failover mode configuration in oak-segment-azure

oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzurePersistenceManager.java

@@ -91,6 +91,14 @@ public static AzurePersistence createAzurePersistenceFrom(Configuration configur
         return createPersistenceFromAccessKey(configuration);
     }
 
+    public static AzurePersistence createAzurePersistenceFromFailover(Configuration configuration) throws IOException {
+        if (!StringUtils.isAnyBlank(configuration.failoverClientId(), configuration.failoverClientSecret(), configuration.failoverTenantId())) {
+            return createPersistenceFromServicePrincipalCredentials(configuration.failoverAccountName(), configuration.failoverContainerName(), configuration.rootPath(), configuration.failoverClientId(), configuration.failoverClientSecret(), configuration.failoverTenantId(), configuration.enableSecondaryLocation(), false);
+        }
+
+        return createPersistenceFromAccessKey(configuration.failoverAccountName(), configuration.failoverContainerName(), configuration.failoverAccessKey(), null, configuration.rootPath(), configuration.enableSecondaryLocation(), false);
+    }
+
     private static AzurePersistence createPersistenceFromAccessKey(Configuration configuration) throws IOException {
         return createPersistenceFromAccessKey(configuration.accountName(), configuration.containerName(), configuration.accessKey(), configuration.blobEndpoint(), configuration.rootPath(), configuration.enableSecondaryLocation(), true);
     }

oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentStoreService.java

@@ -59,26 +59,35 @@ public class AzureSegmentStoreService {
     @Activate
     public void activate(ComponentContext context, Configuration config) throws IOException {
         if (useAzureSdkV12) {
-            log.info("Starting node store using Azure SDK 12");
-            AzurePersistence persistence = AzurePersistenceManager.createAzurePersistenceFrom(config);
-            registration = context.getBundleContext()
-                    .registerService(SegmentNodeStorePersistence.class, persistence, new Hashtable<String, Object>() {{
-                        put(SERVICE_PID, String.format("%s(%s, %s)", AzurePersistence.class.getName(), config.accountName(), config.rootPath()));
-                        if (!Objects.equals(config.role(), "")) {
-                            put("role", config.role());
-                        }
-                    }});
+            AzurePersistence persistence;
+            String accountName;
+            if (config.failoverEnabled()) {
+                log.info("Starting node store using Azure SDK 12 in failover mode");
+                accountName = config.failoverAccountName();
+                persistence = AzurePersistenceManager.createAzurePersistenceFromFailover(config);
+            } else {
+                log.info("Starting node store using Azure SDK 12");
+                accountName = config.accountName();
+                persistence = AzurePersistenceManager.createAzurePersistenceFrom(config);
+            }
+            Hashtable<String, Object> properties = getServiceRegistrationProperties(AzurePersistence.class.getName(), config, accountName);
+
+            registration = context.getBundleContext().registerService(SegmentNodeStorePersistence.class, persistence, properties);
         } else {
             log.info("Starting node store using Azure SDK 8");
+            Hashtable<String, Object> properties = getServiceRegistrationProperties(AzurePersistenceV8.class.getName(), config, config.accountName());
             AzurePersistenceV8 persistence = AzureSegmentStoreV8.createAzurePersistenceFrom(config);
-            registration = context.getBundleContext()
-                    .registerService(SegmentNodeStorePersistence.class, persistence, new Hashtable<String, Object>() {{
-                        put(SERVICE_PID, String.format("%s(%s, %s)", AzurePersistenceV8.class.getName(), config.accountName(), config.rootPath()));
-                        if (!Objects.equals(config.role(), "")) {
-                            put("role", config.role());
-                        }
-                    }});
+            registration = context.getBundleContext().registerService(SegmentNodeStorePersistence.class, persistence, properties);
+        }
+    }
+
+    private Hashtable<String, Object> getServiceRegistrationProperties(String persistenceClassName, Configuration config, String accountName) {
+        Hashtable<String, Object> properties = new Hashtable<>();
+        properties.put(SERVICE_PID, String.format("%s(%s, %s)", persistenceClassName, accountName, config.rootPath()));
+        if (!Objects.equals(config.role(), "")) {
+            properties.put("role", config.role());
         }
+        return properties;
     }
 
     @Deactivate

oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/Configuration.java

@@ -95,4 +95,39 @@
             description = "When set to true specifies that requests will be attempted in primary, then in secondary region." +
                 "Default value is '" + AzureSegmentStoreService.DEFAULT_ENABLE_SECONDARY_LOCATION + "'.")
     boolean enableSecondaryLocation() default AzureSegmentStoreService.DEFAULT_ENABLE_SECONDARY_LOCATION;
+
+    @AttributeDefinition(
+            name = "Enable failover",
+            description = "When set to true, enables failover to the failover Azure Storage account.")
+    boolean failoverEnabled() default false;
+
+    @AttributeDefinition(
+            name = "Azure account name for failover",
+            description = "Name of the Azure Storage account to use for failover.")
+    String failoverAccountName();
+
+    @AttributeDefinition(
+            name = "Azure container name for failover",
+            description = "Name of the Azure Storage container to use for failover.")
+    String failoverContainerName();
+
+    @AttributeDefinition(
+            name = "Azure account access key for failover",
+            description = "Access key which should be used to authenticate on the failover account")
+    String failoverAccessKey();
+
+    @AttributeDefinition(
+            name = "Azure Service Principal ID for failover",
+            description = "Azure Service Principal ID for Azure Storage authentication")
+    String failoverClientId() default "";
+
+    @AttributeDefinition(
+            name = "Azure Service Principal Password for failover",
+            description = "Azure Service Principal Password for Azure Storage authentication")
+    String failoverClientSecret() default "";
+
+    @AttributeDefinition(
+            name = "Azure Active Directory ID for failover",
+            description = "Azure Active Directory ID for Azure Storage authentication")
+    String failoverTenantId() default "";
 }