Use Cluster Linking to Manage Audit Logs on Confluent Cloud

Looking for Confluent Platform Cluster Linking docs? You are currently viewing Confluent Cloud documentation. If you are looking for Confluent Platform docs, check out Cluster Linking on Confluent Platform.

Confluent Audit Logs provide a view into the activity in a Confluent Cloud organization. With Cluster Linking, audit logs can be synced in real time to any Dedicated Confluent Cloud cluster. This enables you to:

  • Consume the audit logs directly from your cluster so that you can reuse your existing consumer configurations and use Stream Governance with the audit logs topic.
  • Manipulate the audit logs and build stream processing applications in ksqlDB.
  • Sync the audit logs into an external data store like Amazon S3, Snowflake, or Splunk with Connect.

Setup

  • Create new or identify any existing Dedicated cluster to serve as the Destination (target) cluster.
  • The audit logs cluster will serve as the Source cluster. Details on how to access this cluster are described in the steps below.
  • Use any cloud provider and networking type. This tutorial uses the public internet, and automatic encryption options.

If you need more guidance on this, see the first steps in the Quick Start for Confluent Cloud about how to create a cluster and choose networking.

Configure Permissions to Consume Audit Logs

Log on to Confluent CLI and select your provisioning environment

confluent environment list
confluent environment use <env id>
Now using "<env id>" as the default (active) environment.

Connect to your cluster as soon as it is available

confluent kafka cluster use <cluster id>
Set Kafka cluster "<cluster id>" as the active cluster for environment "<env id>"

Get the details of your cluster

confluent kafka cluster describe <cluster id>
+------------------+---------------------------------------------------------+
| ID               | <cluster id>                                            |
| Name             | abcdefg-xyz-auditlog-linking                            |
| Type             | DEDICATED                                               |
| Ingress          | 50                                                      |
| Egress           | 150                                                     |
| Storage          | Infinite                                                |
| Provider         | aws                                                     |
| Availability     | single-zone                                             |
| Region           | eu-west-2                                               |
| Status           | UP                                                      |
| Endpoint         | SASL_SSL://pkc-nnnnn.eu-west-2.aws.confluent.cloud:9092 |
| REST Endpoint    | https://pkc-nnnnn.eu-west-2.aws.confluent.cloud:443     |
| Cluster Size     | 1                                                       |
+---------------+------------------------------------------------------------+

Check for Audit Logs

confluent audit-log describe

This will give you the IDs you need to access the audit logs.

+-----------------+-----------------------------+
| Cluster         | <cluster id>                |
| Environment     | <env id>                    |
| Service Account | <sa id>                     |
| Topic Name      | confluent-audit-log-events  |
+-----------------+-----------------------------+

Create a service account using resource IDs and details from previous commands

confluent environment use <env id>
confluent kafka cluster use <cluster id>
confluent api-key create --service-account <service-account-id> --resource <cluster-id>

(Optional) Look up an existing API keys

This is optional. If you already have an API key, you can use the following command to look it up.

confluent api-key list --resource <cluster id>

You will see your keys in the list. If you plan to reuse any of these keys, note that you must know the secret.

Associate your API key with the Audit Log resource

confluent api-key use <api key> --resource <cluster id>
Set API Key "<api key>" as the active API key for "<cluster id>".

Confirm the Audit Log can be consumed

confluent kafka topic consume -b confluent-audit-log-events

Use Ctrl+C to exit the consumer.

Set up Cluster Linking

The Dedicated cluster you created in previous steps will host the mirrored auditlog topic.

In these next steps, you will establish a link on that cluster (the target cluster) and link it to the source (the Audit Log cluster).

  1. Make sure that you are working on the destination cluster (the Dedicated cluster).

    You can use these commands to list environments and clusters, and switch to the appropriate ones.

    confluent environment list
    
    confluent environment use <env id>
    
    confluent kafka cluster list
    
    confluent kafka cluster use <cluster id>
    
  2. Run the following command to create a link called my-link.

    confluent kafka link create my-link --cluster <destination id> \
        --source-cluster-id  <source-id> \
        --source-bootstrap-server <source bootstrap server> \
        --source-api-key <key> --source-api-secret <secret>
    
    • The source-bootstrap-server can be obtained by running the following to get the environment and cluster IDs:

      confluent audit-log describe
      

      Your output will be similar to the following:

      +-----------------+----------------------------+
      | Cluster         | <cluster-id>               |
      | Environment     | env-33r92                  |
      | Service Account | sa-43kzk2                  |
      | Topic Name      | confluent-audit-log-events |
      +-----------------+----------------------------+
      
    • Then run the following commands to locate and describe the Audit Logs source cluster:

      confluent environment use <audit-logs-env-id>
      
      confluent kafka cluster list
      

      Your output should resemble

            Id         |             Name             |   Type   | Provider |  Region   | Availability | Status
      -----------------+------------------------------+----------+----------+-----------+--------------+---------
        * <cluster-id> | _confluent_audit_log_cluster | STANDARD | aws      | us-west-2 | single-zone  | UP
      
    • Run this command to get the details for the Audit Logs source cluster:

      confluent kafka cluster describe <audit-logs-cluster-id>
      

      Your output should resemble:

      +--------------+---------------------------------------------------------+
      | Id           | <cluster-id>                                            |
      | Name         | _confluent_audit_log_cluster                            |
      | Type         | DEDICATED                                               |
      | Ingress      |                                                     100 |
      | Egress       |                                                     100 |
      | Storage      | Infinite                                                |
      | Provider     | aws                                                     |
      | Availability | single-zone                                             |
      | Region       | us-west-2                                               |
      | Status       | UP                                                      |
      | Endpoint     | SASL_SSL://<pkc-id>.us-west-2.aws.confluent.cloud:9092  |
      | RestEndpoint | https://<pkc-id>.us-west-2.aws.confluent.cloud:443      |
      +--------------+---------------------------------------------------------+
      
    • Look under Endpoint for the bootstrap server for your source cluster.

Mirror the Audit Logs Topic and Schema

  1. Click through to view the properties of the link and use the Add mirror topic button to add the mirror. For the topic name, enter confluent-audit-log-events. The link and mirror topic should be displayed, as shown below:

    ../../_images/cluster-links-audit-logs-mirror.png
  2. Import the audit logs topic schema into the Schema Registry for this cluster.

    1. Navigate to the Confluent Cloud Environment that contains your cluster.

    2. Under the Schemas section on the right-hand column, click View and Manage.

      ../../_images/cluster-links-audit-logs-manage-schemas.png
  3. Click Add Schema.

  4. Name the schema: confluent-audit-log-events-value.

    Tip

    Any schema whose name ends with a -value suffix is automatically associated as the value schema for the topic, using the schema name.

  5. Select JSON Schema.

  6. Copy the content from the Audit Log Event Schema into the text box.

    Updated: 04/22/2022

    Following is the complete audit log event schema for Confluent Cloud:

    {
      "$schema":"http://json-schema.org/draft-07/schema#",
      "type":"object",
      "description":"This is v1.2 of the Confluent cloud audit log event schema. It is based on CloudEvents. Links to the latest version can be found at https://docs.confluent.io/cloud/current/monitoring/audit-logging. Confluent will make non-breaking changes to the schema without advance notice. Breaking changes will be widely communicated at least 180 days in advance, and we will continue to maintain compatibility during this time. Exceptions to this policy apply in case of critical security vulnerabilities or functional defects. Check the above web page for links to our announcement channels and details about our compatibility policy.",
      "properties":{
        "id":{
          "type":"string",
          "minLength":1,
          "description":"Uniquely identifies the event.",
          "examples":[
            "c72a3d0c-e6f3-4196-9b49-a835614452df"
          ]
        },
        "specversion":{
          "type":"string",
          "minLength":1,
          "description":"The version of the CloudEvents specification which the event uses.",
          "examples":[
            "1.0"
          ]
        },
        "source":{
          "type":"string",
          "format":"uri-reference",
          "minLength":1,
          "description":"Identifies the context in which an event happened.",
          "examples":[
            "crn://confluent.cloud",
            "crn://confluent.cloud/kafka=lkc-a1b2c",
            "crn:///kafka=lkc-a1b2c"
          ]
        },
        "subject":{
          "type":[
            "string",
            "null"
          ],
          "minLength":1,
          "description":"Identifies the resource that would be affected by the event.",
          "examples":[
            "crn:///kafka=lkc-a1b2c",
            "crn://confluent.cloud/kafka=lkc-a1b2c",
            "crn://confluent.cloud/kafka=lkc-xyz01/topic=departures",
            "crn://confluent.cloud/kafka=lkc-xyz01/group=delivery-estimator"
          ]
        },
        "type":{
          "type":"string",
          "minLength":1,
          "description":"Describes the type of event.",
          "examples":[
            "io.confluent.kafka.server/authentication",
            "io.confluent.kafka.server/authorization",
            "io.confluent.cloud/request"
          ]
        },
        "time":{
          "type":[
            "string",
            "null"
          ],
          "minLength":1,
          "description":"Timestamp of when the occurrence happened. Adheres to RFC 3339.",
          "examples":[
            "2020-12-06T13:39:03Z",
            "2020-12-06T13:39:03.123Z"
          ]
        },
        "datacontenttype":{
          "type":[
            "string",
            "null"
          ],
          "description":"Content type of the data value. Adheres to RFC 2046 format.",
          "minLength":1,
          "examples":[
            "application/json"
          ]
        },
        "dataschema":{
          "type":[
            "string",
            "null"
          ],
          "format":"uri",
          "minLength":1,
          "description":"Identifies the schema that data adheres to. Currently unused."
        },
        "data":{
          "type":"object",
          "description":"Additional details about the audited occurrence.",
          "properties":{
            "serviceName":{
              "type":"string",
              "description":"The resource identifier of the service (the source) that received the request being logged.",
              "examples":[
                "crn://confluent.cloud",
                "crn://confluent.cloud/kafka=lkc-a1b2c",
                "crn:///kafka=lkc-a1b2c"
              ]
            },
            "resourceName":{
              "type":"string",
              "description":"The resource identifier of the target (subject) of the request",
              "examples":[
                "crn:///kafka=lkc-a1b2c",
                "crn://confluent.cloud/kafka=lkc-a1b2c",
                "crn://confluent.cloud/kafka=lkc-xyz01/topic=departures",
                "crn://confluent.cloud/kafka=lkc-xyz01/group=delivery-estimator"
              ]
            },
            "request":{
              "type":[
                "object",
                "null"
              ],
              "description":"Unordered map of dynamically typed values.",
              "additionalProperties":{
                "type":[
                  "array",
                  "boolean",
                  "number",
                  "object",
                  "string",
                  "null"
                ]
              }
            },
            "requestMetadata":{
              "type":[
                "object",
                "null"
              ],
              "description":"Unordered map of dynamically typed values.",
              "additionalProperties":{
                "type":[
                  "array",
                  "boolean",
                  "number",
                  "object",
                  "string",
                  "null"
                ]
              }
            },
            "result":{
              "type":[
                "object",
                "null"
              ],
              "description":"Unordered map of dynamically typed values.",
              "additionalProperties":{
                "type":[
                  "array",
                  "boolean",
                  "number",
                  "object",
                  "string",
                  "null"
                ]
              }
            }
          }
        }
      },
      "required":[
        "id",
        "source",
        "specversion",
        "type"
      ],
      "allOf":[
        {
          "if":{
            "required":[
              "type"
            ],
            "properties":{
              "type":{
                "const":"io.confluent.kafka.server/authentication"
              }
            }
          },
          "then":{
            "properties":{
              "data":{
                "type":"object",
                "description":"Additional details about the authentication check.",
                "properties":{
                  "methodName":{
                    "type":"string",
                    "description":"The type of request being logged.",
                    "examples":[
                      "kafka.Authentication"
                    ]
                  },
                  "authenticationInfo":{
                    "type":"object",
                    "properties":{
                      "principal":{
                        "type":"string",
                        "description":"Identifies the authenticated principal.",
                        "examples":[
                          "User:12345",
                          "None:UNKNOWN_USER"
                        ]
                      },
                      "metadata":{
                        "type":"object",
                        "properties":{
                          "mechanism":{
                            "type":"string",
                            "description":"Authentication mechanism.",
                            "examples":[
                              "SASL_SSL/PLAIN",
                              "SASL_SSL/OAUTHBEARER"
                            ]
                          },
                          "identifier":{
                            "type":"string",
                            "description":"Identifies the numeric user ID or API key supplied by the requester.",
                            "examples":[
                              "MAIDSRFG53RXYTKR",
                              "12345"
                            ]
                          }
                        }
                      }
                    }
                  },
                  "requestMetadata":{
                    "type":"object",
                    "properties":{
                      "client_address":{
                        "type":"string",
                        "description":"Ignore this field until further notice. The address of the client making the request. This field may be missing, or even if present, have an unhelpful value."
                      }
                    }
                  },
                  "result":{
                    "type":"object",
                    "properties":{
                      "status":{
                        "type":"string",
                        "examples":[
                          "SUCCESS",
                          "UNAUTHENTICATED"
                        ]
                      },
                      "message":{
                        "type":"string",
                        "description":"Indicates the result status."
                      }
                    }
                  }
                }
              }
            }
          }
        },
        {
          "if":{
            "required":[
              "type"
            ],
            "properties":{
              "type":{
                "const":"io.confluent.kafka.server/authorization"
              }
            }
          },
          "then":{
            "properties":{
              "data":{
                "type":"object",
                "properties":{
                  "methodName":{
                    "type":"string",
                    "description":"The type of request being logged.",
                    "examples":[
                      "kafka.AlterConfigs",
                      "kafka.CreateAcls",
                      "kafka.CreateTopics",
                      "kafka.DeleteAcls",
                      "kafka.DeleteGroups"
                    ]
                  },
                  "authenticationInfo":{
                    "type":"object",
                    "properties":{
                      "principal":{
                        "type":"string",
                        "description":"Identifies the authenticated principal that made the request.",
                        "examples":[
                          "User:12345"
                        ]
                      }
                    }
                  },
                  "authorizationInfo":{
                    "type":"object",
                    "properties":{
                      "granted":{
                        "type":"boolean",
                        "description":"The result of the authorization check."
                      },
                      "operation":{
                        "$ref":"#/$defs/operation"
                      },
                      "resourceType":{
                        "type":"string",
                        "description":"The type of the resource being checked for authorization.",
                        "examples":[
                          "Cluster",
                          "Group",
                          "Topic"
                        ]
                      },
                      "resourceName":{
                        "type":"string",
                        "description":"The resource name of the checked authorization rule.",
                        "examples":[
                          "kafka-cluster",
                          "delivery-estimator",
                          "departures"
                        ]
                      },
                      "patternType":{
                        "type":"string",
                        "description":"The pattern, LITERAL or PREFIX, used to match the resource against the authorization rule.",
                        "examples":[
                          "LITERAL",
                          "PREFIX"
                        ]
                      },
                      "aclAuthorization":{
                        "type":"object",
                        "description":"Details about an ACL rule.",
                        "properties":{
                          "permissionType":{
                            "$ref":"#/$defs/permissionType"
                          },
                          "host":{
                            "$ref":"#/$defs/host"
                          }
                        }
                      },
                      "rbacAuthorization":{
                        "type":"object",
                        "properties":{
                          "role":{
                            "$ref":"#/$defs/role"
                          },
                          "scope":{
                            "type":"object",
                            "properties":{
                              "outerScope":{
                                "type":"array",
                                "description":"The path elements in the outer scopes with the outermost first",
                                "items":{
                                  "type":"string"
                                }
                              }
                            },
                            "description":"The scope the action was authorized in"
                          }
                        }
                      },
                      "superUserAuthorization":{
                        "$ref":"#/$defs/superUserAuthorization"
                      }
                    }
                  },
                  "request":{
                    "type":"object",
                    "properties":{
                      "clientId":{
                        "type":"string",
                        "description":"This is a user-supplied identifier for the client application. The user can use any identifier they like. This ID acts as a logical grouping across all requests from a particular client.",
                        "examples":[
                          "invoice-processor-admin",
                          "adminclient-42",
                          "the-replicator"
                        ]
                      },
                      "correlationId":{
                        "type":"string",
                        "description":"This is a user-supplied integer. It will be passed back in the response by the server, unmodified. It is useful for matching request and response between the client and server."
                      }
                    }
                  },
                  "requestMetadata":{
                    "type":"object",
                    "properties":{
                      "client_address":{
                        "type":"string",
                        "description":"Ignore this field until further notice. The address of the client making the request. This field may be missing, or even if present, have an unhelpful value."
                      }
                    }
                  }
                }
              }
            }
          }
        },
        {
          "if":{
            "required":[
              "type"
            ],
            "anyOf":[{
              "properties": {
                "type": {
                   "const":"io.confluent.cloud/request"
                }
              }
            },
            {
              "properties": {
                "type": {
                  "const":"io.confluent.kafka.server/request"
                }
              }
            }
          ]
        },
        "then":{
            "properties":{
              "data":{
                "type":"object",
                "description":"Additional details about the cloud request",
                "properties":{
                  "methodName":{
                    "type":"string",
                    "description":"The type of request being logged.",
                    "examples":[
                      "CreateKafkaCluster",
                      "GetAPIKeys",
                      "DeleteEnvironment",
                      "CreateConnector",
                      "DeleteAPIKey"
                    ]
                  },
                  "cloudResources":{
                    "type":"array",
                    "description":"The resource(s) or collection(s) targeted in the operation.",
                    "items":{
                      "type":"object",
                      "properties":{
                        "scope":{
                          "type":"object",
                          "description":"Defines the scope of the action.",
                          "properties":{
                            "resources":{
                              "type":"array",
                              "description":"The list of cloud resources that define the scope of the action.",
                              "items":{
                                "allOf":[{
                                    "$ref":"#/$defs/resource"
                                }]
                              }
                            }
                          }
                        },
                        "resource":{
                          "allOf":[{
                            "$ref":"#/$defs/resource"
                          }]
                        }
                      }
                    }
                  },
                  "resourceName":{
                    "type":"string",
                    "description":"The resource identifier of the target of the request for compatibility with other request types. Will be deprecated in future."
                  },
                  "authenticationInfo":{
                    "type":"object",
                    "description":"Information about the principal and the credentials used to prove its identity.",
                    "properties":{
                      "principal":{
                        "allOf":[{
                          "$ref":"#/$defs/principal",
                          "description":"Identifies the authenticated principal that is used for any authorization checks. Also identifies the authenticated principal that made the request, unless originalPrincipal is set."
                         }]
                      },
                      "originalPrincipal":{
                        "allOf":[{
                          "$ref":"#/$defs/principal",
                          "description":"If set, indicates the original principal that made the request. Used for situations where one principal can assume the identity of a different principal."
                         }]
                      },
                      "result":{
                        "description":"The result of authentication checks on the provided credentials.",
                        "enum":[
                          "UNSET",
                          "SUCCESS",
                          "FAILURE"
                        ]
                      },
                      "errorMessage":{
                        "type":"string",
                        "description":"A short, human-readable description of the reason authentication failed.",
                        "examples":[
                          "Authentication Failed. Wrong Credentials"
                        ]
                      },
                      "credentials":{
                        "allOf":[{
                          "$ref":"#/$defs/credentials"
                        }]
                      }
                    }
                  },
                  "authorizationInfo":{
                    "type":"object",
                    "description":"The result of this authorization check.",
                    "oneOf":[
                    {
                    "properties":{
                      "dryRun":{
                        "type": "boolean"
                      },
                      "result":{
                        "enum":[
                          "UNSET",
                          "ALLOW",
                          "DENY"
                        ]
                      },
                      "operation":{
                        "allOf":[{
                          "$ref":"#/$defs/operation"
                         }]
                      }
                    },
                    "additionalProperties": false
                    },
                      {
                    "properties":{
                        "superUserAuthorization":{
                        "allOf":[{
                          "$ref":"#/$defs/superUserAuthorization"
                          }]
                        }
                      },
                    "additionalProperties": false
                    },
                    {
                    "properties":{
                        "aclAuthorization":{
                          "type":"object",
                          "description":"An ACL rule that resulted in the authorization check being denied or allowed.",
                          "properties":{
                            "permissionType":{
                                "allOf":[{
                                    "$ref":"#/$defs/permissionType"
                                }]
                            },
                            "host":{
                                "allOf":[{
                                    "$ref":"#/$defs/host"
                                }]
                            },
                            "resourceType":{
                                "allOf":[{
                                    "$ref":"#/$defs/resourceType"
                                }]
                            },
                            "patternType":{
                                "allOf":[{
                                    "$ref":"#/$defs/patternType"
                                }]
                            },
                            "patternName":{
                                "allOf":[{
                                    "$ref":"#/$defs/patternName"
                                }]
                            }
                          }
                        }
                      },
                      "additionalProperties":false
                      },
                      {
                      "properties":{
                        "rbacAuthorization":{
                          "type":"object",
                          "description":"A RBAC rule that resulted in the authorization check being allowed.",
                          "properties":{
                            "role":{
                              "allOf":[{
                                "$ref":"#/$defs/role"
                               }]
                            },
                            "cloudScope":{
                              "type":"array",
                              "description":"The list of cloud resources involved in the scope of an action.",
                              "items":{
                                "type":"object",
                                "properties":{
                                  "type":{
                                    "enum":[
                                      "ALL",
                                      "ORGANIZATION",
                                      "ENVIRONMENT",
                                      "CLOUD_CLUSTER",
                                      "USER",
                                      "SERVICE_ACCOUNT",
                                      "API_KEY",
                                      "KAFKA_CLUSTER",
                                      "TOPIC",
                                      "GROUP",
                                      "TRANSACTIONAL_ID",
                                      "SCHEMA_REGISTRY",
                                      "SUBJECT",
                                      "VERSION",
                                      "SCHEMA",
                                      "KSQL_CLUSTER",
                                      "QUERY",
                                      "STREAM",
                                      "TABLE",
                                      "TYPE",
                                      "VARIABLE",
                                      "CONNECT_CLUSTER",
                                      "CONNECTOR",
                                      "SECRET",
                                      "SECURITY_METADATA",
                                      "SSO_CONNECTION",
                                      "USER_INVITATION",
                                      "MARKETPLACE_ENTITLEMENT",
                                      "NETWORK",
                                      "PEERING",
                                      "PRIVATE_LINK_ACCESS",
                                      "TRANSIT_GATEWAY_ATTACHMENT",
                                      "PIPELINE"
                                    ]
                                  },
                                  "resourceId":{
                                    "type":"string",
                                    "description":"Resource Identifier",
                                    "examples":[
                                      "sa-8191a",
                                      "lkc-a1b2c"
                                    ]
                                  }
                                }
                              }
                            },
                            "resourceType":{
                              "allOf":[{
                                "$ref":"#/$defs/resourceType"
                              }]
                            },
                            "patternType":{
                              "allOf":[{
                                "$ref":"#/$defs/patternType"
                              }]
                            },
                            "patternName":{
                              "allOf":[{
                                 "$ref":"#/$defs/patternName"
                              }]
                            }
                          }
                        }
                      },
                      "additionalProperties": false
                    }
                    ]
                  },
                  "requestMetadata":{
                    "type":"object",
                    "properties":{
                      "connectionId":{
                        "type":"string",
                        "description":"Uniquely identifies an authenticated connection. Only present for successfully authenticated connections."
                      },
                      "requestId":{
                        "type":"array",
                        "description":"Uniquely identifies a client request.",
                        "items":{
                          "type":"string"
                        }
                      },
                      "clientAddress":{
                        "type":"array",
                        "description":"Network address of the remote client.",
                        "items":{
                          "type":"object",
                          "properties":{
                            "ip":{
                              "type":"string",
                              "description":"IPv4 or IPv6 address.",
                              "anyOf":[
                                {
                                  "format":"ipv4"
                                },
                                {
                                  "format":"ipv6"
                                }
                              ]
                            },
                            "port":{
                              "type":"number",
                              "description":"Port number, if known."
                            }
                          }
                        }
                      },
                      "clientId":{
                        "type":"string",
                        "description":"A client-provided identifier, logged for correlation, as a courtesy to the client."
                      },
                      "clientTraceId":{
                        "type":"string",
                        "description":"A client-provided identifier, logged for correlation, as a courtesy to the client."
                      }
                    }
                  },
                  "request":{
                    "type":"object",
                    "description":"Describes additional audit-worthy details about the request.",
                    "properties":{
                      "accessType":{
                        "description":"The type of the request",
                        "enum":[
                          "UNKNOWN",
                          "READ_ONLY",
                          "MODIFICATION"
                        ]
                      },
                      "data":{
                        "type":[
                          "object",
                          "null"
                        ],
                        "description":"Audit-worthy details from the request. There are currently no guarantees of backwards-compatibility on the contents of this field. The contents of this field should not be relied on for any programmatic access.",
                        "additionalProperties":{
                          "type":[
                            "array",
                            "boolean",
                            "number",
                            "object",
                            "string",
                            "null"
                          ]
                        }
                      }
                    }
                  },
                  "result":{
                    "type":"object",
                    "description":"Describes the result of the overall operation (i.e. success, failure, etc) along with any audit-worthy details. For example, this field contains the resource identifier of a newly created resource from a CREATE request.",
                    "properties":{
                      "status":{
                        "description":"The status of the result.",
                        "enum":[
                          "UNSET",
                          "SUCCESS",
                          "FAILURE"
                        ]
                      },
                      "data":{
                        "type":[
                          "object",
                          "null"
                        ],
                        "description":"Audit-worthy details from the operation's result. There are currently no guarantees of backwards-compatibility on the contents of this field. The contents of this field should not be relied on for any programmatic access.",
                        "additionalProperties":{
                          "type":[
                            "array",
                            "boolean",
                            "number",
                            "object",
                            "string",
                            "null"
                          ]
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        }
      ],
      "$defs":{
        "user":{
          "type":"object",
          "properties":{
            "resourceId":{
              "type":"string",
              "description":"Resource identifier",
              "examples":[
                "u-8191a"
              ]
            }
          }
        },
        "operation":{
          "type":"string",
          "description":"Identifies the operation being checked for authorization.",
          "examples":[
            "Alter",
            "AlterConfigs",
            "Create",
            "Delete",
            "DescribeConfigs"
          ]
        },
        "resourceType":{
          "type":"string",
          "description":"The type of the resource being checked for authorization.",
          "examples":[
            "Cluster",
            "Group",
            "Topic"
          ]
        },
        "resourceName":{
          "type":"string",
          "description":"The resource name of the checked authorization rule.",
          "examples":[
            "kafka-cluster",
            "delivery-estimator",
            "departures"
          ]
        },
        "patternType":{
          "type":"string",
          "description":"The pattern, LITERAL or PREFIX, used to match the resource against the authorization rule.",
          "examples":[
            "LITERAL",
            "PREFIX"
          ]
        },
        "patternName":{
          "type":"string",
          "description":"The resource name or prefix matched by this ACL rule.",
          "examples":[
            "*"
          ]
        },
        "permissionType":{
          "type":"string",
          "examples":[
            "ALLOW",
            "DENY"
          ]
        },
        "host":{
          "type":"string",
          "description":"Host to which the ACL rule applies, usually wildcard (*).",
          "examples":[
            "*"
          ]
        },
        "role":{
          "type":"string",
          "description":"A role that the principal has, granting authorization to perform the operation on the resource.",
          "examples":[
            "OrganizationAdmin",
            "EnvironmentAdmin",
            "CloudClusterAdmin",
            "MetricsViewer"
          ]
        },
        "resource":{
          "type":"object",
          "description":"The actual resource affected within the identified scope.",
          "properties":{
            "type":{
              "enum":[
                "ALL",
                "ORGANIZATION",
                "ENVIRONMENT",
                "CLOUD_CLUSTER",
                "USER",
                "SERVICE_ACCOUNT",
                "API_KEY",
                "KAFKA_CLUSTER",
                "TOPIC",
                "GROUP",
                "TRANSACTIONAL_ID",
                "SCHEMA_REGISTRY",
                "SUBJECT",
                "VERSION",
                "SCHEMA",
                "KSQL_CLUSTER",
                "QUERY",
                "STREAM",
                "TABLE",
                "TYPE",
                "VARIABLE",
                "CONNECT_CLUSTER",
                "CONNECTOR",
                "SECRET",
                "SECURITY_METADATA",
                "SSO_CONNECTION",
                "USER_INVITATION",
                "MARKETPLACE_ENTITLEMENT",
                "NETWORK",
                "PEERING",
                "PRIVATE_LINK_ACCESS",
                "TRANSIT_GATEWAY_ATTACHMENT",
                "PIPELINE",
                "CLUSTER_LINK"
              ]
            },
            "resourceId":{
              "type":"string",
              "description":"Resource Identifier",
              "examples":[
                "sa-8191a",
                "lkc-a1b2c"
              ]
            }
          }
        },
        "principal":{
          "type":"object",
          "properties":{
            "email":{
              "type":"string",
              "description":"Email address of the principal.",
              "examples":[
                "abc@confluent.io"
              ]
            },
            "confluentUser":{
              "allOf":[{
                "$ref": "#/$defs/user"
              }]
            },
            "confluentServiceAccount":{
              "allOf":[{
                "$ref": "#/$defs/user"
              }]
            },
            "externalAccount":{
              "type":"object",
              "properties":{
                "namespace":{
                  "type":"array",
                  "items":{
                    "type":"object",
                    "properties":{
                      "type":{
                        "type":"string",
                        "examples":[
                          "idp",
                          "tenant"
                        ]
                      },
                      "id":{
                        "type":"string",
                        "examples":[
                          "confluent.cloud",
                          "a2efbdb2-af88-4b38-99a4-9a1e9d700963"
                        ]
                      }
                    }
                  }
                },
                "subject":{
                  "type":"string",
                  "description":"Identity of the requesting party, known to the IdP.",
                  "examples":[
                    "bilbo.baggins",
                    "harry.potter"
                  ]
                }
              }
            }
          },
          "oneOf":[
            {
              "required":[
                "confluentServiceAccount"
              ]
            },
            {
              "required":[
                "confluentUser"
              ]
            },
            {
              "required":[
                "externalAccount"
              ]
            }
          ]
        },
        "credentials":{
          "type":"object",
          "description":"Information sent to prove the identity of the principal.",
          "oneOf":[
            {
            "properties":{
              "idSecretCredentials":{
                "type":"object",
                "properties":{
                  "credentialId":{
                    "type":"string",
                    "description":"Identifies the credential",
                    "examples":[
                      "bilbo.baggins",
                      "harry.potter"
                    ]
                  }
                }
              },
              "mechanism":{
                "allOf":[{
                    "$ref": "#/$defs/mechanism"
                    }]
                }
              },
              "additionalProperties": false
            },
            {
            "properties":{
              "idTokenCredentials":{
                "type":"object",
                "properties":{
                  "type":{
                    "type":"string",
                    "description":"The type of the Id Token Credential. JWT for the foreseeable future.",
                    "examples":[
                      "JWT"
                    ]
                  },
                  "issuer":{
                    "type":"string",
                    "description":"Who signed the token."
                  },
                  "subject":{
                    "type":"string",
                    "description":"Identifies the principal."
                  },
                  "audience":{
                    "type":"array",
                    "items":{
                      "type":"string"
                    }
                  }
                }
              },
              "mechanism":{
                "allOf":[{
                    "$ref": "#/$defs/mechanism"
                }]
              }
            },
            "additionalProperties": false
            },
            {
            "properties":{
              "certificateCredentials":{
                "type":"object",
                "properties":{
                  "dname":{
                    "type":"object",
                    "description":"The principal identified by this certificate.",
                    "properties":{
                      "cn":{
                        "type":"string",
                        "description":"Common name"
                      },
                      "ou":{
                        "type":"string",
                        "description":"Organizational unit"
                      },
                      "o":{
                        "type":"string",
                        "description":"Organization"
                      },
                      "l":{
                        "type":"string",
                        "description":"Locality"
                      },
                      "st":{
                        "type":"string",
                        "description":"State or province"
                      },
                      "c":{
                        "type":"string",
                        "description":"Country"
                      }
                    }
                  }
                }
              },
            "mechanism":{
                "allOf":[{
                    "$ref": "#/$defs/mechanism"
                    }]
                }
            },
            "additionalProperties": false
            },
            {
            "properties":{
              "delegateCredentials":{
                "type":"object",
                "properties":{
                  "delegatePrincipal":{
                    "type":"object",
                    "description":"Identifies the authenticated principal that made the request.",
                    "oneOf":[
                    {
                    "properties":{
                          "email":{
                            "type":"string",
                            "description":"Email address of the principal.",
                            "examples":[
                              "abc@confluent.io"
                            ]
                          },
                          "confluentUser":{
                            "type":"object",
                            "properties":{
                              "resourceId":{
                                "type":"string",
                                "description":"Resource identifier",
                                "examples":[
                                  "u-8191a"
                                ]
                              }
                            }
                          }
                        },
                        "additionalProperties": false
                     },
                     {
                     "properties":{
                         "email":{
                           "type": "string",
                           "description": "Email address of the principal.",
                           "examples":[
                            "abc@confluent.io"
                          ]
                         },
                          "confluentServiceAccount":{
                            "type":"object",
                            "properties":{
                              "resourceId":{
                                "type":"string",
                                "description":"Resource identifier",
                                "examples":[
                                  "u-8191a"
                                ]
                              }
                            }
                          }
                        },
                        "additionalProperties": false
                        },
                        {
                        "properties":{
                          "email":{
                            "type":"string",
                            "description":"Email address of the principal.",
                            "examples":[
                               "abc@confluent.io"
                                ]
                            },
                          "externalAccount":{
                            "type":"object",
                            "properties":{
                              "namespace":{
                                "type":"array",
                                "items":{
                                  "type":"object",
                                  "properties":{
                                    "type":{
                                      "type":"string",
                                      "examples":[
                                        "idp",
                                        "tenant"
                                      ]
                                    },
                                    "id":{
                                      "type":"string",
                                      "examples":[
                                        "confluent.cloud",
                                        "a2efbdb2-af88-4b38-99a4-9a1e9d700963"
                                      ]
                                    }
                                  }
                                }
                              },
                              "subject":{
                                "type":"string",
                                "description":"Identity of the requesting party, known to the IdP.",
                                "examples":[
                                  "bilbo.baggins",
                                  "harry.potter"
                                ]
                              }
                            }
                          }
                        },
                        "additionalProperties": false
                        }
                      ]
                    },
                  "delegateCredentials":{
                    "allOf":[{
                    "$ref":"#/$defs/credentials"
                  }]
                }
              }
            },
            "mechanism":{
              "allOf":[{
                "$ref": "#/$defs/mechanism"
                }]
              }
            },
            "additionalProperties": false
          }]
        },
        "mechanism":{
          "enum":[
            "UNSET",
            "SASL_PLAIN",
            "SASL_SCRAM",
            "SASL_OAUTHBEARER",
            "SASL_GSSAPI",
            "MTLS",
            "HTTP_BASIC",
            "HTTP_BEARER"
          ]
        },
        "superUserAuthorization":{
          "type":"boolean",
          "description":"If true, access was authorized because principal is a super-user."
        }
      }
    }
    
  7. Save the schema.

View the Mirror Topic from the Dedicated Cluster

Follow these steps to view the Audit Logs mirror topic on the dedicated cluster (destination).

  1. In Confluent Cloud, select your environment and your dedicated cluster instance (the URI for this resource will be something like confluent.cloud/environments/your_env_id/clusters/your_cluster_id/).

  2. Click through to view the messages on your dedicated cluster.

    ../../_images/cluster-links-audit-logs-view-mirrot-topic-messages-on-dest.png
  3. Click on the See in Stream Lineage in the top right to see which consumer groups are consuming the audit logs topic.

    ../../_images/cluster-links-audit-log-consumers-stream-lineage.png

Suggested Resources