Add konflux_utils module to monitor Konflux Pipelineruns

CLOUDDST-28645

Signed-off-by: Yashvardhan Nanavati <[email protected]>

Assisted-by: Cursor
Signed-off-by: Yashvardhan Nanavati <[email protected]>

Author: yashvardhannanavati

README.md

@@ -446,6 +446,15 @@ The custom configuration options for the Celery workers are listed below:
 * `iib_ocp_opm_mapping` - the dictionary mapping of OCP version to OPM version
 indicating the OPM version to be used for the corresponding OCP version like
 `{"v4.15": "opm-v1.28.0"}`
+* `iib_konflux_cluster_url` - the URL of the Konflux OpenShift cluster to access for Tekton PipelineRuns
+  (e.g. `https://api.konflux.example.com:6443`). This is required for cross-cluster access to Konflux.
+* `iib_konflux_cluster_token` - the authentication token for accessing the Konflux OpenShift cluster.
+  This should be a service account token with appropriate permissions to access Tekton PipelineRuns.
+* `iib_konflux_cluster_ca_cert` - the CA certificate for the Konflux OpenShift cluster. This can be
+  either a file path to the certificate or the certificate content as a string. This is required
+  for secure cross-cluster access.
+* `iib_konflux_namespace` - the namespace in the Konflux cluster where Tekton PipelineRuns are located.
+  This defaults to `iib-tenant`.
 
 
 If you wish to configure AWS S3 bucket for storing artifact files, the following **environment variables**

iib/workers/config.py

@@ -127,6 +127,12 @@ class Config(object):
     # The minimal version of OPM which requires setting the --migrate-level flag for migrate
     iib_opm_new_migrate_version = "v1.46.0"
 
+    # Konflux configuration for cross-cluster access
+    iib_konflux_cluster_url: Optional[str] = None
+    iib_konflux_cluster_token: Optional[str] = None
+    iib_konflux_cluster_ca_cert: Optional[str] = None
+    iib_konflux_namespace: str = 'iib-tenant'
+
 
 class ProductionConfig(Config):
     """The production IIB Celery configuration."""
@@ -326,6 +332,7 @@ def validate_celery_config(conf: app.utils.Settings, **kwargs) -> None:
 
     _validate_multiple_opm_mapping(conf['iib_ocp_opm_mapping'])
     _validate_iib_org_customizations(conf['iib_organization_customizations'])
+    _validate_konflux_config(conf)
 
     if conf.get('iib_aws_s3_bucket_name'):
         if not isinstance(conf['iib_aws_s3_bucket_name'], str):
@@ -481,6 +488,46 @@ def _validate_iib_org_customizations(
                         )
 
 
+def _validate_konflux_config(conf: app.utils.Settings) -> None:
+    """
+    Validate Konflux configuration variables.
+
+    :param celery.app.utils.Settings conf: the Celery application configuration to validate
+    :raises iib.exceptions.ConfigError: if the configuration is invalid
+    """
+    konflux_url = conf.get('iib_konflux_cluster_url')
+    konflux_token = conf.get('iib_konflux_cluster_token')
+    konflux_ca_cert = conf.get('iib_konflux_cluster_ca_cert')
+
+    if any([konflux_url, konflux_token, konflux_ca_cert]):
+        _validate_konflux_required_fields(konflux_url, konflux_token, konflux_ca_cert)
+        _validate_konflux_field_types(konflux_url, konflux_token, konflux_ca_cert)
+
+
+def _validate_konflux_required_fields(konflux_url, konflux_token, konflux_ca_cert):
+    """Validate that all required Konflux fields are provided."""
+    if not konflux_url:
+        raise ConfigError('iib_konflux_cluster_url must be set when using Konflux configuration')
+    if not konflux_token:
+        raise ConfigError('iib_konflux_cluster_token must be set when using Konflux configuration')
+    if not konflux_ca_cert:
+        raise ConfigError(
+            'iib_konflux_cluster_ca_cert must be set when using Konflux configuration'
+        )
+
+
+def _validate_konflux_field_types(konflux_url, konflux_token, konflux_ca_cert):
+    """Validate the types and formats of Konflux configuration fields."""
+    if not isinstance(konflux_url, str) or not konflux_url.startswith('https://'):
+        raise ConfigError('iib_konflux_cluster_url must be a valid HTTPS URL')
+
+    if not isinstance(konflux_token, str):
+        raise ConfigError('iib_konflux_cluster_token must be a string')
+
+    if not isinstance(konflux_ca_cert, str):
+        raise ConfigError('iib_konflux_cluster_ca_cert must be a string')
+
+
 def get_worker_config() -> app.utils.Settings:
     """Return the Celery configuration."""
     # Import this here to avoid a circular import