def _apply_labels_to_deployment_config()

in cli/src/klio_cli/commands/job/gke.py [0:0]

• 57 lines of code
• 15 McCabe index (conditional complexity)

    def _apply_labels_to_deployment_config(self, deployment_config):
        # `metadata.labels` are a best practices thing, but not required
        # (these would be "deployment labels"). At least one label defined in
        # `spec.template.metadata.labels` is required for k8s deployments
        # ("pod labels").
        # There also must be at least one "selector label"
        # (`spec.selector.matchLabels`) which connects the deployment to pod.
        # More info: https://stackoverflow.com/a/54854179
        # TODO: add environment labels if/when we support dev/test/prod envs
        metadata_labels, pod_labels, selector_labels = {}, {}, {}

        # standard practice labels ("app" and "role")
        existing_metadata_labels = glom.glom(
            deployment_config, "metadata.labels", default={}
        )
        metadata_app = glom.glom(existing_metadata_labels, "app", default=None)
        if not metadata_app:
            job_name = self.klio_config.job_name
            metadata_labels["app"] = job_name
        metadata_labels.update(existing_metadata_labels)

        existing_pod_labels = glom.glom(
            deployment_config, "spec.template.metadata.labels", default={}
        )
        pod_app = glom.glom(existing_pod_labels, "app", default=None)
        pod_role = glom.glom(existing_pod_labels, "role", default=None)
        if not pod_app:
            pod_app = metadata_labels["app"]
        if not pod_role:
            # just drop hyphens from `app` value
            pod_role = "".join(pod_app.split("-"))
        pod_labels["app"] = pod_app
        pod_labels["role"] = pod_role
        pod_labels.update(existing_pod_labels)

        existing_selector_labels = glom.glom(
            deployment_config, "spec.selector.matchLabels", default={}
        )
        selector_app = glom.glom(existing_selector_labels, "app", default=None)
        selector_role = glom.glom(
            existing_selector_labels, "role", default=None
        )
        if not selector_app:
            selector_labels["app"] = pod_labels["app"]
        if not selector_role:
            selector_labels["role"] = pod_labels["role"]
        selector_labels.update(existing_selector_labels)

        # klio-specific labels
        pod_labels["klio/klio_cli_version"] = klio_cli_version

        # deployment labels
        deploy_user = os.environ.get("USER", "unknown")
        if os.environ.get("CI", "").lower() == "true":
            deploy_user = "ci"
        pod_labels["klio/deployed_by"] = deploy_user

        # any user labels from klio_config.pipeline_options
        # note: if pipeline_options.label (singular) is define in
        # klio-job.yaml, klio-core appends it to pipeline_options.labels
        # (plural) automatically
        user_labels_list = self.klio_config.pipeline_options.labels
        # user labels in beam/klio config are lists of strings, where the
        # strings are key=value pairs, e.g. "keyfoo=valuebar"
        user_labels = {}
        for user_label in user_labels_list:
            if "=" not in user_label:
                # skip - not a valid label; this should technically be
                # caught when validating configuration (not yet implemented)
                continue

            # theoretically user_label could be key=value1=value2, so
            # we just take the first one, but value1=value2 is not
            # valid and will be caught during validation below.
            key, value = user_label.split("=", 1)
            user_labels[key] = value
        pod_labels.update(user_labels)

        path_to_labels = (
            ("metadata.labels", metadata_labels),
            ("spec.selector.matchLabels", selector_labels),
            ("spec.template.metadata.labels", pod_labels),
        )
        for label_path, label_dict in path_to_labels:
            # raises if not valid
            RunPipelineGKE._validate_labels(label_path, label_dict)
            glom.assign(
                deployment_config, label_path, label_dict, missing=dict
            )