Configure Unified Stream Manager

The Unified Stream Manager Agent acts as a centralized proxy and gateway for Apache Kafka®. Deploy Confluent Unified Stream Manager (USM) with Confluent for Kubernetes (CFK) to connect customer-managed Confluent Platform clusters to Confluent Cloud and enable Confluent Cloud features for them. CFK deploys and manages the Unified Stream Manager Agent in your Kubernetes environment.

This topic covers the steps for deploying Unified Stream Manager in Confluent Platform using CFK. Deploying the Unified Stream Manager Agent is part of the Register your Confluent Platform Kafka cluster in Confluent Cloud workflow. Review the steps in that topic before you deploy Unified Stream Manager.

Requirements and considerations

Deploying Unified Stream Manager with Confluent for Kubernetes requires:

Ensure your Kubernetes environment is set up and CFK is installed.

Use the appropriate version that supports your Confluent Platform and Unified Stream Manager requirements.
Unified Stream Manager Agent to Confluent Cloud communication only supports basic authentication.
Unified Stream Manager Agent to Confluent Platform components communication supports three authentication combinations: basic authentication, basic authentication with TLS, and mutual TLS (mTLS) authentication. Basic authentication and mTLS cannot be combined.

Configure and deploy Unified Stream Manager

Configure Unified Stream Manager Agent using the USMAgent custom resource (CR), and then apply the CR using the kubectl apply command.
```
kind: USMAgent
spec:
  replicas:
  image:
    application:              --- [1]
    init:                     --- [2]
  authentication:
    type:                     --- [3]
    basic:
      secretRef:              --- [4]
  tls:
    secretRef:                --- [5]
  confluentCloudClient:
    endpoint:                 --- [6]
    environmentId:            --- [7]
    authentication:
      type:                   --- [8]
      basic:
        secretRef:            --- [9]
  externalAccess:             --- [10]
    type:                     --- [11]
    loadBalancer:             --- [12]
    nodePort:                 --- [13]
```
- [1] Set to the Unified Stream Manager application image.
- [2] Set to the Unified Stream Manager CFK init container image.
- [3] Set to basic or mtls.
- [4] Required for basic authentication. Specify the secret containing the basic authentication credentials.
- [5] For TLS between Unified Stream Manager Agent and Confluent Platform components, specify the secret containing the TLS certificate, the key, and the root certificate authority (CA) files.
- [6] Specify the Confluent Cloud endpoint.
  
  The Confluent Cloud endpoint is available in the output file generated when you perform the first step in the registration process. See Generate the configuration file.
  
  Complete this step before you deploy the Unified Stream Manager Agent.
- [7] Specify the Confluent Cloud Environment ID.
  
  The Environment ID is available in the output file generated when you perform the first step in the registration process. See Generate the configuration file.
  
  Complete this step before you deploy the Unified Stream Manager Agent.
- [8] Set to basic for basic authentication.
- [9] Required for basic authentication. Specify the secret containing the Cloud API key and secret. The values are available in the output file generated when you perform the first step in the registration process. See Generate the configuration file.
  
  Complete this step before you deploy the Unified Stream Manager Agent.
- [10] Optional. External access is not required for the Unified Stream Manager Agent.
- [11] Set to loadBalancer, nodePort, or route to specify the external access type. For OpenShift support, see USM support for OpenShift.
- [12] Required when externalAccess type ([11]) is set to loadBalancer. For configuring load balancers, see Configure Load Balancers.
- [13] Required when externalAccess type ([11]) is set to nodePort. For configuring node ports, see Configure Node Ports.
Configure the client-side properties in Kafka, KRaft, and Connect for communication with Unified Stream Manager Agent, and then apply the changes to the CRs with the kubectl apply command.
```
spec:
  dependencies:
    usmAgentClient:
      url:                    --- [1]
      clientMetricsEnabled:   --- [2]
      authentication:
        type:                 --- [3]
        basic:
          secretRef:                  --- [4]
          directoryPathInContainer:   --- [5]
      tls:
        enabled:                      --- [6]
        secretRef:                    --- [7]
        directoryPathInContainer:     --- [8]
```
- [1] Specify the Unified Stream Manager Agent URL.
- [2] Optional. Set to true to enable client metrics. Supported in CFK 3.2.x with Confluent Platform 7.9.6 or later, or 8.1.2 or later.
- [3] Set to basic or mtls to specify the authentication type.
  
  For credential setup and the required format, see Basic authentication credentials.
- [4] Specify the secret containing the basic authentication credentials.
- [5] Specify the basic authentication credential secret path in the container. For details, see Provide secrets in HashiCorp Vault.
- [6] Set to true or false to enable or disable TLS.
- [7] Specify the secret containing the TLS certificate.
- [8] Specify the TLS certificate secret path in the container. For details, see Provide secrets in HashiCorp Vault.
Register your Confluent Platform Connect cluster in Confluent Cloud.

You can use the following options to retrieve the Connect cluster ID (also known as the group ID) that is required to register the Connect cluster in Confluent Cloud:
- Use the kubectl describe connect command, and fetch the group ID under the Status section.
- If you have the Confluent CLI installed, you can use the command confluent connect cluster list as described in Register your Confluent Platform Connect cluster in Confluent Cloud.

Configure the Command Handler for connector management

Starting in CFK 3.4.0, you can enable the Command Handler on the USM Agent to manage self-managed Connect connectors from Confluent Cloud.

When you set spec.commandHandler on the USMAgent CR, CFK runs an extra container, usm-agent-command-handler, in the USM Agent pod. The Command Handler receives connector management commands from Confluent Cloud, such as create, update, pause, resume, restart, and delete. It runs them against the REST APIs of your Connect clusters.

The Command Handler is opt-in. Without spec.commandHandler, the USM Agent runs as before in monitoring mode.

Prerequisites

Before you enable the Command Handler, ensure you have the following:

CFK 3.4.0 with the 3.4.0 init container.
USM Agent 1.3.0.
A Connect cluster registered in USM and connected to a Confluent Cloud Kafka cluster (LKC). For the registration steps, see Register your Confluent Platform Connect cluster in Confluent Cloud. Connector management from Confluent Cloud is an Early Access feature and supports only Confluent Platform Connect clusters.

Configure the Command Handler

Add spec.commandHandler to the USMAgent CR, and then apply the CR using the kubectl apply command.

In the simplest configuration, reference the Connect CR by name. CFK discovers the endpoint and group ID from the Connect CR.

kind: USMAgent
spec:
  commandHandler:
    connectClusters:
      - connectClusterRef:
          name: connect

To override the discovered settings or to set authentication and TLS, specify the additional fields. The following example shows all the available settings.

kind: USMAgent
spec:
  commandHandler:
    configOverrides:
      server:                       --- [1]
        - confluent.usm-agent.command-handler.poll.interval.ms=6000
    connectClusters:                --- [2]
      - connectClusterRef:
          name:                     --- [3]
          namespace:                --- [4]
        groupId:                    --- [5]
        connectRest:
          endpoint:                 --- [6]
          authentication:
            type:                   --- [7]
            basic:
              secretRef:            --- [8]
          tls:
            secretRef:              --- [9]

[1] Optional. Overrides the Command Handler server properties.
[2] Specify one entry for each Connect cluster to manage. You can specify more than one cluster. Each connectClusterRef must be unique.
[3] Required. The name of the Connect CR.
[4] Optional. The namespace of the Connect CR. Defaults to the namespace of the USM Agent.
[5] Optional. The group ID of the Connect cluster. If you leave it out, CFK reads it from the status of the Connect CR. CFK leaves out a cluster that is not discovered yet and adds it once it is discovered.
[6] Optional. The REST endpoint of the Connect cluster. If you leave it out, CFK reads it from the status of the Connect CR. If you set connectRest.tls, the endpoint must use https.
[7] Optional. Leave out authentication when the Connect REST API requires no authentication. If you set it, basic is the only supported type.
[8] Required for basic authentication. The secret that contains the basic authentication credentials. The secret uses a basic.txt file with the lines username=<user> and password=<password>.
[9] Optional. Set to use TLS. The secret that contains the PEM CA that signed the Connect certificate, under ca.crt or cacerts.pem.

The connectRest block is optional. Leave it out when the Connect REST API requires no authentication or TLS.

To pass custom flags to the Command Handler, set USM_AGENT_COMMAND_HANDLER_CUSTOM_ARGS in spec.podTemplate.envVars. For example, set it to -log-level debug to enable debug log mode.

Command Handler behavior

Be aware of the following behavior:

Any change to spec.commandHandler rolls the USM Agent pod. A change to a secret that it references, such as a credential or CA rotation, also rolls the pod.
When you remove spec.commandHandler, the pod rolls back to the USM Agent alone, in monitoring mode.

Validation errors

The USMAgent CR fails with an event in the following cases:

connectRest.authentication.type is mtls, bearer, or oauth. Only basic is supported.
groupId is blank.
connectClusterRef has duplicate entries.
connectRest.tls is set with an http endpoint.
The CA secret has no PEM CA. An example is a Java KeyStore (JKS) only secret, such as an auto-generated *-generated-jks secret. Use a PEM CA secret instead.

Troubleshoot the Command Handler

Use the following to troubleshoot the Command Handler:

Health: The Command Handler serves /healthz on port 9930. It serves the status of each cluster on /statusz on port 9920. Port 9920 is loopback-only, so use kubectl port-forward to reach it.

Logs: Check the Command Handler container logs.

kubectl logs <usm_agent_pod> -c usm-agent-command-handler

HTTP traffic: Capture the Command Handler command polls and status reports, together with metrics and events.
```
kubectl confluent cluster usmagent logcollector --req-type all
```

USM support for OpenShift

Confluent for Kubernetes supports deploying Unified Stream Manager Agent on OpenShift clusters. This section walks through deploying the Unified Stream Manager Agent on OpenShift end to end, including configuring external access using OpenShift routes, which provide a way to expose your services outside the OpenShift cluster.

Prerequisites

An OpenShift cluster
CFK installed on the cluster

Deploy USM Agent on OpenShift using routes

OpenShift routes require TLS to be enabled. The Unified Stream Manager Agent is accessible at https://<prefix>.<domain>:443. For example, https://usm-agent.apps.example.com:443.

Set up namespace and security policies

Create the namespace:
```
oc create namespace confluent
```

Configure security context constraints:

oc adm policy add-scc-to-user anyuid -z default -n confluent
oc adm policy add-scc-to-user anyuid -z confluent-for-kubernetes -n confluent

Create TLS certificates and secrets

You can create custom certificates (for production) or use auto-generated certificates (for testing).

For custom certificates:

Create a certificate directory and generate the CA:

mkdir -p ~/usm-certs && cd ~/usm-certs
openssl genrsa -out ca-key.pem 2048
openssl req -x509 -new -nodes -key ca-key.pem -days 365 -out ca.pem -subj "/CN=USMAgentCA/O=Confluent"

Create a Subject Alternative Name (SAN) configuration file. Replace apps.example.com with your OpenShift domain:

cat > usm-agent-san.cnf << 'EOF'
[req]
distinguished_name = req_distinguished_name
req_extensions = v3_req
prompt = no
[req_distinguished_name]
CN = usm-agent
[v3_req]
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
subjectAltName = @alt_names
[alt_names]
DNS.1 = usm-agent
DNS.2 = usm-agent.confluent
DNS.3 = usm-agent.confluent.svc
DNS.4 = usm-agent.confluent.svc.cluster.local
DNS.5 = *.apps.example.com
DNS.6 = usm-agent.apps.example.com
EOF

Generate the server certificate:

openssl genrsa -out usm-agent-key.pem 2048
openssl req -new -key usm-agent-key.pem -out usm-agent.csr -config usm-agent-san.cnf
openssl x509 -req -in usm-agent.csr -CA ca.pem -CAkey ca-key.pem -CAcreateserial \
  -out usm-agent.pem -days 365 -extensions v3_req -extfile usm-agent-san.cnf

Create a Java KeyStore (JKS) truststore:

keytool -importcert -alias usm-agent-ca -file ca.pem -keystore truststore.jks \
  -storepass changeit -noprompt

Create secrets:

oc -n confluent create secret tls usm-agent-tls \
  --cert=$HOME/usm-certs/usm-agent.pem \
  --key=$HOME/usm-certs/usm-agent-key.pem

oc -n confluent patch secret usm-agent-tls \
  -p "{\"data\":{\"ca.crt\":\"$(base64 < $HOME/usm-certs/ca.pem | tr -d '\n')\"}}"

oc -n confluent create secret generic usm-ca-trust \
  --from-file=truststore.jks=$HOME/usm-certs/truststore.jks \
  --from-literal=jksPassword.txt='jksPassword=changeit'

Verify secrets:

oc -n confluent get secrets | grep -E "usm-agent-tls|usm-ca-trust"

Example output:

usm-agent-tls   kubernetes.io/tls     3      10s
usm-ca-trust    Opaque                2      5s

For auto-generated certificates, set spec.tls.autoGeneratedCerts: true and omit secretRef in the custom resource.

Deploy USM Agent

Create a Unified Stream Manager Agent CR YAML file:

These are some of the configuration fields:

autoGeneratedCerts: Set to false for custom certificates. Set to true for auto-generated.
secretRef: The secret name. If you are using custom certificates, set this to usm-agent-tls.
domain: Your OpenShift cluster domain. For example, apps.example.com.
prefix: The route prefix. The default is usm-agent.
wildcardPolicy: Subdomain or None. The default is None.
annotations: OpenShift route annotations. This is optional.

Here is an example configuration:

apiVersion: platform.confluent.io/v1beta1
kind: USMAgent
metadata:
  name: usm-agent
  namespace: confluent
spec:
  tls:
    autoGeneratedCerts: false
    secretRef: usm-agent-tls
  externalAccess:
    type: route
    route:
      domain: apps.example.com
      prefix: usm-agent
      annotations:
        haproxy.router.openshift.io/timeout: 5500ms
      wildcardPolicy: None
  replicas: 1
  image:
    application: confluentinc/cp-usm-agent:latest
    init: confluentinc/confluent-init-container:latest
  confluentCloudClient:
    endpoint: https://api.confluent.cloud:443
    environmentId: env-xxxxx
    authentication:
      type: basic
      basic:
        secretRef: ccloud-credentials

Apply the configuration:
```
oc apply -f <usm_agent_cr>
```

Get the OpenShift router external IP:

oc get svc --namespace openshift-ingress

Example output:

NAME                      TYPE           CLUSTER-IP       EXTERNAL-IP    PORT(S)                      AGE
router-default            LoadBalancer   172.30.84.52     20.189.181.8   80:31294/TCP,443:32145/TCP   42h

Get the route hostname:

oc get routes | grep usm-agent

Example output:

NAME                  HOST/PORT                              PATH   SERVICES    PORT       TERMINATION   WILDCARD
usm-agent-bootstrap   usm-agent.apps.example.com                    usm-agent   external   passthrough   None

Add the DNS entry in your DNS provider mapping the external IP to the route hostname:
```
20.189.181.8 usm-agent.apps.example.com
```

Test the Unified Stream Manager Agent endpoint:

ROUTE_HOST=$(oc get route usm-agent-bootstrap -o jsonpath='{.spec.host}')
curl -k "https://${ROUTE_HOST}"

For additional validation steps, see Validate route connections to Confluent Platform components.