Module ploigos_step_runner.step_implementers.deploy.argocd_deploy

StepImplementer for the deploy step using ArgoCD.

Step Configuration

Step configuration expected as input to this step. Could come from:

• static configuration
• runtime configuration
• previous step results

[container-image-pull-repository, container-image-push-repository, container-image-repository] | Yes | | Container image repository within the given container image registry to pull container image from when deploying. [container-image-pull-digest, container-image-push-digest, container-image-digest] | Maybe | | Ifuse-container-image-digestis <code>True</code>, container image digest to pull container image with when deploying.[container-image-pull-tag, container-image-push-tag, container-image-tag] | Maybe | | If use-container-image-digest is False, container image tag to pull container image with when deploying. use-container-image-short-addres | No | False | If True then use container image short address (no registry). If False then use container image full address (including registry). use-container-image-digest | No | True | If True then use container image digest in container image address when pulling container image for deployment.
If False then use container image tag in container image address when pulling container image for deployment. organization | Yes | | Used to build ArgoCD application name. application-name | Yes | | Used to build ArgoCD application name. service-name | Yes | | Used to build ArgoCD application name. branch | Yes | | Used to build ArgoCD application name.

Results

Results output by this step.

Classes

class ArgoCDDeploy (workflow_result, parent_work_dir_path, config, environment=None)

StepImplementer for the deploy step using ArgoCD.

Expand source code

class ArgoCDDeploy(ContainerDeployMixin, ArgoCDGeneric):
    """`StepImplementer` for the `deploy` step using ArgoCD.
    """

    @staticmethod
    def step_implementer_config_defaults():
        """Getter for the StepImplementer's configuration defaults.

        Returns
        -------
        dict
            Default values to use for step configuration values.

        Notes
        -----
        These are the lowest precedence configuration values.
        """
        return {**ContainerDeployMixin.step_implementer_config_defaults(), **DEFAULT_CONFIG}

    @staticmethod
    def _required_config_or_result_keys():
        """Getter for step configuration or previous step result artifacts that are required before
        running this step.

        See Also
        --------
        _validate_required_config_or_previous_step_result_artifact_keys

        Returns
        -------
        array_list
            Array of configuration keys or previous step result artifacts
            that are required before running the step.
        """
        return REQUIRED_CONFIG_OR_PREVIOUS_STEP_RESULT_ARTIFACT_KEYS + \
            ContainerDeployMixin._required_config_or_result_keys()

    def _validate_required_config_or_previous_step_result_artifact_keys(self):
        """Validates that the required configuration keys or previous step result artifacts
        are set and have valid values.

        Validates that:
        * required configuration is given
        * either both git-username and git-password are set or neither.

        Raises
        ------
        StepRunnerException
            If step configuration or previous step result artifacts have invalid required values
        """
        super()._validate_required_config_or_previous_step_result_artifact_keys()

        # ensure either both git-username and git-password are set or neither
        runtime_auth_config = {}
        for auth_config_key in GIT_AUTHENTICATION_CONFIG:
            runtime_auth_config_value = self.get_value(auth_config_key)

            if runtime_auth_config_value is not None:
                runtime_auth_config[auth_config_key] = runtime_auth_config_value

        if (any(element in runtime_auth_config for element in GIT_AUTHENTICATION_CONFIG)) and \
                (not all(element in runtime_auth_config for element in GIT_AUTHENTICATION_CONFIG)):
            raise StepRunnerException(
                "Either 'git-username' or 'git-password 'is not set. Neither or both must be set."
            )

        # ensure if deployment config repo is using http/https that user/pass is provided
        deployment_config_repo_uri = self.get_value('deployment-config-repo')
        if re.match(r'^http://|^https://', deployment_config_repo_uri) and not runtime_auth_config:
            raise StepRunnerException(
                f"Since provided 'deployment-config-repo' ({deployment_config_repo_uri}) uses"
                f" http/https protical both 'git-username' and 'git-password' must be provided."
            )

    def _run_step(self):  # pylint: disable=too-many-locals, too-many-statements
        """Runs the step implemented by this StepImplementer.

        Returns
        -------
        StepResult
            Object containing the dictionary results of this step.
        """
        step_result = StepResult.from_step_implementer(self)

        # get input
        deployment_config_repo = self.get_value('deployment-config-repo')
        deployment_config_repo_branch = self._get_repo_branch()
        deployment_config_helm_chart_path = self.get_value('deployment-config-helm-chart-path')
        deployment_config_destination_cluster_uri = self.get_value('kube-api-uri')
        deployment_config_destination_cluster_token = self.get_value('kube-api-token')
        deployment_config_helm_chart_environment_values_file = \
            self._get_deployment_config_helm_chart_environment_values_file()
        deployment_config_helm_chart_values_file_container_image_address_yq_path = self.get_value([
            'deployment-config-helm-chart-values-file-container-image-address-yq-path',
            'deployment-config-helm-chart-values-file-image-tag-yq-path'
        ])
        deployment_config_helm_chart_additional_value_files = \
            self.get_value('deployment-config-helm-chart-additional-values-files')
        force_push_tags = self.get_value('force-push-tags')
        additional_helm_values_files = self.get_value('additional-helm-values-files')

        try:
            argocd_app_name = self._get_app_name()
            step_result.add_artifact(
                name='argocd-app-name',
                value=argocd_app_name
            )

            # clone the configuration repository
            print("Clone the configuration repository")
            deployment_config_repo_dir = self.create_working_dir_sub_dir('deployment-config-repo')

            clone_repo(
                repo_dir=deployment_config_repo_dir,
                repo_url=deployment_config_repo,
                username = self.get_value('git-username'),
                password = self.get_value('git-password')
            )
            git_config(
                repo_dir=deployment_config_repo_dir,
                git_email=self.get_value('git-email'),
                git_name=self.get_value('git-name'),
            )
            git_checkout(
                repo_dir=deployment_config_repo_dir,
                repo_branch=deployment_config_repo_branch,
            )

            # update values file, commit it, push it, and tag it
            print("Update the environment values file")
            deployment_config_helm_chart_environment_values_file_path = os.path.join(
                deployment_config_repo_dir,
                deployment_config_helm_chart_path,
                deployment_config_helm_chart_environment_values_file
            )
            container_image_address = self._get_deploy_time_container_image_address()
            self._update_yaml_file_value(
                file=deployment_config_helm_chart_environment_values_file_path,
                yq_path=deployment_config_helm_chart_values_file_container_image_address_yq_path,
                value=container_image_address
            )
            step_result.add_artifact(
                name='container-image-deployed-address',
                value=container_image_address,
                description='Container image address used to deploy the container.'
            )

            print("Commit the updated environment values file")
            git_commit_file(
                git_commit_message=f'Updating values for deployment to {self.environment}',
                file_path=os.path.join(
                    deployment_config_helm_chart_path,
                    deployment_config_helm_chart_environment_values_file
                ),
                repo_dir=deployment_config_repo_dir
            )

            print("Tag and push the updated environment values file")
            deployment_config_repo_tag = self._get_deployment_config_repo_tag()
            self._git_tag_and_push_deployment_config_repo(
                deployment_config_repo=deployment_config_repo,
                deployment_config_repo_dir=deployment_config_repo_dir,
                deployment_config_repo_tag=deployment_config_repo_tag,
                force_push_tags=force_push_tags
            )
            step_result.add_artifact(
                name='config-repo-git-tag',
                value=deployment_config_repo_tag
            )

            # create/update argocd app and sync it
            print("Sign into ArgoCD")
            self._argocd_sign_in(
                argocd_api=self.get_value('argocd-api'),
                username=self.get_value('argocd-username'),
                password=self.get_value('argocd-password'),
                insecure=self.get_value('argocd-skip-tls')
            )

            add_or_update_target_cluster = self.get_value('argocd-add-or-update-target-cluster')
            if add_or_update_target_cluster:
                print("Add target cluster to ArgoCD")
                self._argocd_add_target_cluster(
                    kube_api=deployment_config_destination_cluster_uri,
                    kube_api_token=deployment_config_destination_cluster_token,
                    kube_api_skip_tls=self.get_value('kube-api-skip-tls')
                )

            print("Determine deployment namespace")
            deployment_namespace = self.get_value('deployment-namespace')
            if deployment_namespace:
                print(f"  Using user provided namespace name: {deployment_namespace}")
            else:
                deployment_namespace = argocd_app_name
                print(f"  Using auto generated namespace name: {deployment_namespace}")

            print(f"Create or update ArgoCD Application ({argocd_app_name})")
            argocd_values_files = []
            argocd_values_files += deployment_config_helm_chart_additional_value_files
            argocd_values_files += [deployment_config_helm_chart_environment_values_file]
            argocd_values_files += additional_helm_values_files
            self._argocd_app_create_or_update(
                argocd_app_name=argocd_app_name,
                repo=deployment_config_repo,
                revision=deployment_config_repo_tag,
                path=deployment_config_helm_chart_path,
                dest_server=deployment_config_destination_cluster_uri,
                dest_namespace=deployment_namespace,
                auto_sync=self.get_value('argocd-auto-sync'),
                fail_on_shared_resource=self.get_value('argocd-fail-on-shared-resource'),
                values_files=argocd_values_files,
                project=self.get_value('argocd-project')
            )

            # sync and wait for the sync of the ArgoCD app
            print(f"Sync (and wait for) ArgoCD Application ({argocd_app_name})")
            self._argocd_app_sync(
                argocd_app_name=argocd_app_name,
                argocd_sync_timeout_seconds=self.get_value('argocd-sync-timeout-seconds'),
                argocd_sync_retry_limit=self.get_value('argocd-sync-retry-limit'),
                argocd_sync_prune=self.get_value('argocd-sync-prune')
            )

            # get the ArgoCD app manifest that was synced
            print(f"Get ArgoCD Application ({argocd_app_name}) synced manifest")
            argocd_app_manifest_file = self._argocd_get_app_manifest(
                argocd_app_name=argocd_app_name
            )
            step_result.add_artifact(
                name='argocd-deployed-manifest',
                value=argocd_app_manifest_file
            )

            # determine the deployed host URLs
            print(
                "Determine the deployed host URLs for the synced"
                f" ArgoCD Application (({argocd_app_name})"
            )
            deployed_host_urls = self._get_deployed_host_urls(
                manifest_path=argocd_app_manifest_file
            )
            step_result.add_artifact(
                name='deployed-host-urls',
                value=deployed_host_urls
            )
        except StepRunnerException as error:
            step_result.success = False
            step_result.message = f"Error deploying to environment ({self.environment}):" \
                                  f" {str(error)}"

        return step_result

Ancestors

• ContainerDeployMixin
• ArgoCDGeneric
• StepImplementer
• abc.ABC

Subclasses

• ArgoCD

Inherited members

• ContainerDeployMixin:
  • step_implementer_config_defaults
• ArgoCDGeneric:
  • argocd_api
  • argocd_password
  • argocd_skip_tls
  • argocd_username
  • config
  • create_working_dir_sub_dir
  • environment
  • get_config_value
  • get_copy_of_runtime_step_config
  • get_result_value
  • get_value
  • global_config_defaults
  • global_environment_config_defaults
  • has_config_value
  • run_step
  • step_config
  • step_config_overrides
  • step_environment_config
  • step_name
  • sub_step_implementer_name
  • sub_step_name
  • work_dir_path
  • workflow_result
  • write_working_file