opensearch-project · mikaylathompson · Jun 25, 2024 · Jun 20, 2024 · Jun 20, 2024 · Jun 20, 2024
@@ -5,7 +5,7 @@ COPY python/Pipfile python/Pipfile.lock ./
 RUN apt -y update
 RUN apt -y install python3 python3-pip
 RUN pip3 install pipenv
-RUN pipenv install --system --deploy --ignore-pipfile
+RUN pipenv install --deploy --ignore-pipfile
 
 
 ENV FM_CODE_PATH /code

@@ -58,6 +58,9 @@ public static class Args {
         @Parameter(names = {"--target-password"}, description = "Optional.  The target password; if not provided, will assume no auth on target", required = false)
         public String targetPass = null;
 
+        @Parameter(names = {"--target-insecure"}, description = "Allow untrusted SSL certificates for target", required = false)
+        public boolean targetInsecure = false;
+
         @Parameter(names = {"--index-allowlist"}, description = ("Optional.  List of index names to migrate"
             + " (e.g. 'logs_2024_01, logs_2024_02').  Default: all indices"), required = false)
         public List<String> indexAllowlist = List.of();
@@ -103,11 +106,14 @@ public static void main(String[] args) throws Exception {
         final String targetHost = arguments.targetHost;
         final String targetUser = arguments.targetUser;
         final String targetPass = arguments.targetPass;
+        final boolean targetInsecure = arguments.targetInsecure;
         final List<String> indexTemplateAllowlist = arguments.indexTemplateAllowlist;
         final List<String> componentTemplateAllowlist = arguments.componentTemplateAllowlist;
         final int awarenessDimensionality = arguments.minNumberOfReplicas + 1;
 
-        final ConnectionDetails targetConnection = new ConnectionDetails(targetHost, targetUser, targetPass);
+        final ConnectionDetails targetConnection = new ConnectionDetails(targetHost, targetUser, targetPass, targetInsecure);
+
+
 
 
 

@@ -1,20 +1,22 @@
 from console_api.apps.orchestrator.serializers import OpenSearchIngestionCreateRequestSerializer
 from console_link.models.osi_utils import (InvalidAuthParameters, create_pipeline_from_json, start_pipeline,
                                            stop_pipeline)
-from console_link.models.migration import MigrationType
 from rest_framework.decorators import api_view, parser_classes
 from rest_framework.parsers import JSONParser
 from rest_framework.response import Response
 from rest_framework import status
 from pathlib import Path
 import boto3
 import datetime
+from enum import Enum
 import logging
 
 logger = logging.getLogger(__name__)
 
 PIPELINE_TEMPLATE_PATH = f"{Path(__file__).parents[4]}/osiPipelineTemplate.yaml"
 
+MigrationType = Enum('MigrationType', ['OSI_HISTORICAL_MIGRATION'])
+
 
 def pretty_request(request, data):
     headers = ''

@@ -5,6 +5,7 @@
 import console_link.logic.metrics as logic_metrics
 import console_link.logic.backfill as logic_backfill
 import console_link.logic.snapshot as logic_snapshot
+import console_link.logic.metadata as logic_metadata
 
 from console_link.models.utils import ExitCode
 from console_link.environment import Environment
@@ -36,10 +37,10 @@
 @click.option('-v', '--verbose', count=True, help="Verbosity level. Default is warn, -v is info, -vv is debug.")
 @click.pass_context
 def cli(ctx, config_file, json, verbose):
-    ctx.obj = Context(config_file)
-    ctx.obj.json = json
     logging.basicConfig(level=logging.WARN - (10 * verbose))
     logger.info(f"Logging set to {logging.getLevelName(logger.getEffectiveLevel())}")
+    ctx.obj = Context(config_file)
+    ctx.obj.json = json
 
 
 # ##################### CLUSTERS ###################
@@ -238,6 +239,26 @@
 # ##################### METRICS ###################
 
 
+@cli.group(name="metadata")
+@click.pass_obj
+def metadata_group(ctx):
+    """All actions related to metadata migration"""
+    if ctx.env.metadata is None:
+        raise click.UsageError("Metadata is not set")
+
+
+@metadata_group.command(name="migrate")
+@click.option("--detach", is_flag=True, help="Run metadata migration in detached mode")
+@click.pass_obj
+def migrate_metadata_cmd(ctx, detach):
+    exitcode, message = logic_metadata.migrate(ctx.env.metadata, detach)
+    if exitcode != ExitCode.SUCCESS:
+        raise click.ClickException(message)
+    click.echo(message)
+
+# ##################### METRICS ###################
+
+
 @cli.group(name="metrics")
 @click.pass_obj
 def metrics_group(ctx):

@@ -9,6 +9,7 @@
 import yaml
 from cerberus import Validator
 
+from console_link.models.metadata import Metadata
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,7 @@ def get_snapshot(config: Dict, source_cluster: Cluster, target_cluster: Cluster)
     "backfill": {"type": "dict", "required": False},
     "metrics_source": {"type": "dict", "required": False},
     "snapshot": {"type": "dict", "required": False},
+    "metadata_migration": {"type": "dict", "required": False}
 }
 
 
@@ -35,11 +37,14 @@ class Environment:
     backfill: Optional[Backfill] = None
     metrics_source: Optional[MetricsSource] = None
     snapshot: Optional[Snapshot] = None
+    metadata: Optional[Metadata] = None
 
     def __init__(self, config_file: str):
+        logger.info(f"Loading config file: {config_file}")
         self.config_file = config_file
         with open(self.config_file) as f:
             self.config = yaml.safe_load(f)
+            logger.info(f"Loaded config file: {self.config}")
         v = Validator(SCHEMA)
         if not v.validate(self.config):
             logger.error(f"Config file validation errors: {v.errors}")
@@ -79,3 +84,7 @@ def __init__(self, config_file: str):
             logger.info(f"Snapshot initialized: {self.snapshot}")
         else:
             logger.info("No snapshot provided")
+        if 'metadata_migration' in self.config:
+            self.metadata: Metadata = Metadata(self.config["metadata_migration"],
+                                               target_cluster=self.target_cluster,
+                                               snapshot=self.snapshot)
@@ -0,0 +1,22 @@
+from typing import Tuple
+
+from console_link.models.metadata import Metadata
+from console_link.models.utils import ExitCode, generate_log_file_path
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def migrate(metadata: Metadata, detached: bool) -> Tuple[ExitCode, str]:
+    logger.info("Migrating metadata")
+    if detached:
+        log_file = generate_log_file_path("metadata_migration")
+        logger.info(f"Running in detached mode, writing logs to {log_file}")
+    try:
+        result = metadata.migrate(detached_log=log_file if detached else None)
+    except Exception as e:
+        logger.error(f"Failed to migrate metadata: {e}")
+        return ExitCode.FAILURE, f"Failure when migrating metadata: {e}"
+    if result.success:
+        return ExitCode.SUCCESS, result.value
+    return ExitCode.FAILURE, result.value
@@ -67,7 +67,7 @@ class Cluster:
     aws_secret_arn: Optional[str] = None
     auth_type: Optional[AuthMethod] = None
     auth_details: Optional[Dict[str, Any]] = None
-    allow_insecure: bool = None
+    allow_insecure: bool = False
 
     def __init__(self, config: Dict) -> None:
         logger.info(f"Initializing cluster with config: {config}")
@@ -128,6 +128,7 @@ def execute_benchmark_workload(self, workload: str,
         elif self.auth_type == AuthMethod.SIGV4:
             raise NotImplementedError(f"Auth type {self.auth_type} is not currently support for executing "
                                       f"benchmark workloads")
+        # Note -- we should censor the password when logging this command
         logger.info(f"Running opensearch-benchmark with '{workload}' workload")
         subprocess.run(f"opensearch-benchmark execute-test --distribution-version=1.0.0 "
                        f"--target-host={self.endpoint} --workload={workload} --pipeline=benchmark-only --test-mode "

@@ -0,0 +1,205 @@
+import os
+import subprocess
+from typing import Optional
+from cerberus import Validator
+import tempfile
+import logging
+
+from console_link.models.command_result import CommandResult
+from console_link.models.schema_tools import list_schema
+from console_link.models.cluster import AuthMethod, Cluster
+from console_link.models.snapshot import S3Snapshot, Snapshot, FileSystemSnapshot
+
+logger = logging.getLogger(__name__)
+
+FROM_SNAPSHOT_SCHEMA = {
+    "type": "dict",
+    # In the future, there should be a "from_snapshot" and a "from_live_cluster" option, but for now only snapshot is
+    # supported, so this is required. It _can_ be null, but that requires a snapshot to defined on its own in
+    # the services.yaml file.
+    "required": True,
+    "nullable": True,
+    "schema": {
+        "snapshot_name": {"type": "string", "required": True},
+        "local_dir": {"type": "string", "required": False},
+        "s3": {
+            'type': 'dict',
+            "required": False,
+            'schema': {
+                'repo_uri': {'type': 'string', 'required': True},
+                'aws_region': {'type': 'string', 'required': True},
+            }
+        },
+        "fs": {
+            'type': 'dict',
+            "required": False,
+            'schema': {
+                'repo_path': {'type': 'string', 'required': True},
+            }
+        }
+    },
+    # We _should_ have the check below, but I need to figure out how to combine it with a potentially
+    # nullable block (like this one)
+    # 'check_with': contains_one_of({'s3', 'fs'})
+
+}
+
+SCHEMA = {
+    "from_snapshot": FROM_SNAPSHOT_SCHEMA,
+    "min_replicas": {"type": "integer", "min": 0, "required": False},
+    "index_allowlist": list_schema(required=False),
+    "index_template_allowlist": list_schema(required=False),
+    "component_template_allowlist": list_schema(required=False)
+}
+
+
+def generate_tmp_dir(name: str) -> str:
+    return tempfile.mkdtemp(prefix=f"migration-{name}-")
+
+
+class Metadata:
+    def __init__(self, config, target_cluster: Cluster, snapshot: Optional[Snapshot] = None):
+        logger.debug(f"Initializing Metadata with config: {config}")
+        v = Validator(SCHEMA)
+        if not v.validate(config):
+            logger.error(f"Invalid config: {v.errors}")
+            raise ValueError(v.errors)
+        self._config = config
+        self._target_cluster = target_cluster
+        self._snapshot = snapshot
+
+        if (not snapshot) and (config["from_snapshot"] is None):
+            raise ValueError("No snapshot is specified or can be assumed "
+                             "for the metadata migration to use.")
+
+        self._min_replicas = config.get("min_replicas", 0)
+        self._index_allowlist = config.get("index_allowlist", None)
+        self._index_template_allowlist = config.get("index_template_allowlist", None)
+        self._component_template_allowlist = config.get("component_template_allowlist", None)
+        logger.debug(f"Min replicas: {self._min_replicas}")
+        logger.debug(f"Index allowlist: {self._index_allowlist}")
+        logger.debug(f"Index template allowlist: {self._index_template_allowlist}")
+        logger.debug(f"Component template allowlist: {self._component_template_allowlist}")
+
+        # If `from_snapshot` is fully specified, use those values to define snapshot params
+        if config["from_snapshot"] is not None:
+            logger.debug("Using fully specified snapshot config")
+            self._init_from_config()
+        else:
+            logger.debug("Using independently specified snapshot")
+            if isinstance(snapshot, S3Snapshot):
+                self._init_from_s3_snapshot(snapshot)
+            elif isinstance(snapshot, FileSystemSnapshot):
+                self._init_from_fs_snapshot(snapshot)
+
+        if config["from_snapshot"] is not None and "local_dir" in config["from_snapshot"]:
+            self._local_dir = config["from_snapshot"]["local_dir"]
+        else:
+            self._local_dir = generate_tmp_dir(self._snapshot_name)
+
+        logger.debug(f"Snapshot name: {self._snapshot_name}")
+        if self._snapshot_location == 's3':
+            logger.debug(f"S3 URI: {self._s3_uri}")
+            logger.debug(f"AWS region: {self._aws_region}")
+        else:
+            logger.debug(f"Local dir: {self._local_dir}")
+
+        logger.info("Metadata migration configuration defined")
+
+    def _init_from_config(self) -> None:
+        config = self._config
+        self._snapshot_location = 's3' if 's3' in config["from_snapshot"] else 'fs'
+        self._snapshot_name = config["from_snapshot"]["snapshot_name"]
+
+        if self._snapshot_location == 'fs':
+            self._repo_path = config["from_snapshot"]["fs"]["repo_path"]
+        else:
+            self._s3_uri = config["from_snapshot"]["s3"]["repo_uri"]
+            self._aws_region = config["from_snapshot"]["s3"]["aws_region"]
+
+    def _init_from_s3_snapshot(self, snapshot: S3Snapshot) -> None:
+        self._snapshot_name = snapshot.snapshot_name
+        self._snapshot_location = "s3"
+        self._s3_uri = snapshot.s3_repo_uri
+        self._aws_region = snapshot.s3_region
+
+    def _init_from_fs_snapshot(self, snapshot: FileSystemSnapshot) -> None:
+        self._snapshot_name = snapshot.snapshot_name
+        self._snapshot_location = "fs"
+        self._repo_path = snapshot.repo_path
+
+    def migrate(self, detached_log=None) -> CommandResult:
+        password_field_index = None
+        command = [
+            "/root/metadataMigration/bin/MetadataMigration",
+            # Initially populate only the required params
+            "--snapshot-name", self._snapshot_name,
+            "--target-host", self._target_cluster.endpoint,
+            "--min-replicas", str(self._min_replicas)
+        ]
+        if self._snapshot_location == 's3':
+            command.extend([
+                "--s3-local-dir", self._local_dir,
+                "--s3-repo-uri", self._s3_uri,
+                "--s3-region", self._aws_region,
+            ])
+        elif self._snapshot_location == 'fs':
+            command.extend([
+                "--file-system-repo-path", self._repo_path,
+            ])
+
+        if self._target_cluster.auth_details == AuthMethod.BASIC_AUTH:
+            try:
+                command.extend([
+                    "--target-username", self._target_cluster.auth_details.get("username"),
+                    "--target-password", self._target_cluster.auth_details.get("password")
+                ])
+                password_field_index = len(command) - 1
+                logger.info("Using basic auth for target cluster")
+            except KeyError as e:
+                raise ValueError(f"Missing required auth details for target cluster: {e}")
+
+        if self._target_cluster.allow_insecure:
+            command.append("--target-insecure")
+
+        if self._index_allowlist:
+            command.extend(["--index-allowlist", ",".join(self._index_allowlist)])
+
+        if self._index_template_allowlist:
+            command.extend(["--index-template-allowlist", ",".join(self._index_template_allowlist)])
+
+        if self._component_template_allowlist:
+            command.extend(["--component-template-allowlist", ",".join(self._component_template_allowlist)])
+
+        if password_field_index:
+            display_command = command[:password_field_index] + ["********"] + command[password_field_index:]
+        else:
+            display_command = command
+        logger.info(f"Migrating metadata with command: {' '.join(display_command)}")
+
+        if detached_log:
+            return self._run_as_detached_process(command, detached_log)
+        return self._run_as_synchronous_process(command)
+
+    def _run_as_synchronous_process(self, command) -> CommandResult:
+        try:
+            # Pass None to stdout and stderr to not capture output and show in terminal
+            subprocess.run(command, stdout=None, stderr=None, text=True, check=True)
+            logger.info(f"Metadata migration for snapshot {self._snapshot_name} completed")
+            return CommandResult(success=True, value="Metadata migration completed")
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Failed to migrate metadata: {str(e)}")
+            return CommandResult(success=False, value=f"Failed to migrate metadata: {str(e)}")
+
+    def _run_as_detached_process(self, command, log_file) -> CommandResult:
+        try:
+            with open(log_file, "w") as f:
+                # Start the process in detached mode
+                process = subprocess.Popen(command, stdout=f, stderr=subprocess.STDOUT, preexec_fn=os.setpgrp)
+                logger.info(f"Metadata migration process started with PID {process.pid}")
+                logger.info(f"Metadata migration logs available at {log_file}")
+                return CommandResult(success=True, value=f"Metadata migration started with PID {process.pid}\n"
+                                     f"Logs are being written to {log_file}")
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Failed to create snapshot: {str(e)}")
+            return CommandResult(success=False, value=f"Failed to migrate metadata: {str(e)}")