Use Cluster Linking to Manage Audit Logs on Confluent Cloud¶
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 |
| Cloud | 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 an API key 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).
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>
Run the following command to create a link called
my-link.confluent kafka link create my-link --cluster <destination id> \ --source-cluster <source-id> \ --source-bootstrap-server <source bootstrap server> \ --source-api-key <key> --source-api-secret <secret>
Tip
--source-cluster-idwas replaced with--source-clusterin version 3 of confluent CLI, as described in the command reference for confluent kafka link create.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-abc123 | | 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 | Cloud | 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 | | Cloud | 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
Endpointfor the bootstrap server for your source cluster.
View the cluster link in Confluent Cloud¶
To see the cluster link listed with the source (listed as external) and the destination, go to https://confluent.cloud/cluster-links.
Mirror the audit logs topic and schema¶
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:
Import the audit logs topic schema into the Schema Registry for this cluster.
Navigate to the Confluent Cloud Environment that contains your cluster.
Under the Schemas section on the right-hand column, click View and Manage.
Click Add Schema.
Name the schema:
confluent-audit-log-events-value.Tip
Any schema whose name ends with a
-valuesuffix is automatically associated as the value schema for the topic, using the schema name.Select JSON Schema.
Copy the content from the Audit Log Event Schema on Confluent Cloud into the text box.
Updated: October 4, 2023
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" ] } } }, "principalResourceId":{ "type":"string", "description":"Authenticated principal resourceId of the requester.", "examples":[ "u-yw9507" ] }, "identity":{ "type":"string", "description":"Identity of the requester in the format.", "examples":[ "crn://confluent.cloud/organization=uuid-for-ourcorp/identity-provider=ourcorp-idp/identity=u-yw9507" ] } } }, "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." }, "connection_id": { "type":"string", "description":"Uniquely identifies an authenticated connection.Only present for successfully authenticated connections.May be used during analysis to correlate events", "examples":[ "111222686238900021" ] }, "network_id": { "type":"string", "description":"Uniquely identifies an authenticated network.", "examples":[ "n-ab1324" ] } } }, "result":{ "type":"object", "properties":{ "status":{ "type":"string", "examples":[ "SUCCESS", "UNAUTHENTICATED" ] }, "message":{ "type":"string", "description":"Indicates the result status." } } }, "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." } } } } } } } } }, { "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" ] }, "identity":{ "type":"string", "description":"Identity of the requester in the format.", "examples":[ "crn://confluent.cloud/organization=uuid-for-ourcorp/identity-provider=ourcorp-idp/identity=u-yw9507" ] } } }, "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" }, "actingPrincipal": { "type": "string", "description": "The decisive principal used for authorization." } } }, "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" }, "actingPrincipal": { "type": "string", "description": "The decisive principal used for authorization." } } }, "superUserAuthorization":{ "$ref":"#/$defs/superUserAuthorization" }, "assignedPrincipals": { "type": "array", "description": "The list of principals the user has access to, including user itself and identity pools", "items": { "type": "string" } } } }, "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." } } }, "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." } } } }, "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.cloud/authorization" } } }, { "properties": { "type": { "const":"io.confluent.kafka.server/request" } } }, { "properties": { "type": { "const":"io.confluent.ksql.server/authentication" } } }, { "properties": { "type": { "const":"io.confluent.ksql.server/authorization" } } }, { "properties": { "type": { "const":"io.confluent.sg.server/authentication" } } }, { "properties": { "type": { "const":"io.confluent.sg.server/authorization" } } } ] }, "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" }] } }, "identity":{ "type":"string", "description":"Identity of the requester in the format.", "examples":[ "crn://confluent.cloud/organization=uuid-for-ourcorp/identity-provider=ourcorp-idp/identity=u-yw9507" ] } }, "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" }] }, "actingPrincipal": { "allOf": [{ "$ref": "#/$defs/principal", "description": "The decisive principal used for authorization." }] } } }, "assignedPrincipals": { "type": "array", "description": "The list of principals the user has access to, including user itself and identity pools", "items": { "allOf": [{ "$ref": "#/$defs/principal" }] } } }, "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": "object", "description":"The list of cloud resources involved in the scope of an action.", "properties": { "resources": { "type":"array", "description":"The list of cloud resources that define the scope of the 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", "DNS_FORWARDER", "SECRET", "SECURITY_METADATA", "SSO_CONNECTION", "USER_INVITATION", "MARKETPLACE_ENTITLEMENT", "NETWORK", "PEERING", "PRIVATE_LINK_ACCESS", "TRANSIT_GATEWAY_ATTACHMENT", "PIPELINE", "CLUSTER_LINK", "IDENTITY_PROVIDER", "IDENTITY_POOL", "CUSTOM_CONNECTOR_PLUGIN", "CUSTOM_CONNECTOR_PLUGIN_VERSION", "FLINK_CLUSTER", "COMPUTE_POOL", "FLINK_REGION", "FLINK_WORKSPACE", "STATEMENT", "NETWORK_LINK_ENDPOINT", "NETWORK_LINK_SERVICE", "NETWORK_LINK_SERVICE_ASSOCIATION", "NS_NOTIFICATION_TYPE", "NS_SUBSCRIPTION", "NS_INTEGRATION", "NS_TEMPLATE", "NS_NOTIFICATION", "PRIVATE_LINK_ATTACHMENT", "PRIVATE_LINK_ATTACHMENT_CONNECTION" ] }, "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" }] }, "actingPrincipal": { "allOf": [{ "$ref": "#/$defs/principal", "description": "The decisive principal used for authorization." }] } } }, "assignedPrincipals": { "type": "array", "description": "The list of principals the user has access to, including user itself and identity pools", "items": { "allOf": [{ "$ref": "#/$defs/principal" }] } } } } ] }, "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", "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", "IDENTITY_PROVIDER", "IDENTITY_POOL", "CUSTOM_CONNECTOR_PLUGIN", "CUSTOM_CONNECTOR_PLUGIN_VERSION", "FLINK_CLUSTER", "FLINK_REGION", "FLINK_WORKSPACE", "STATEMENT", "NETWORK_LINK_ENDPOINT", "NETWORK_LINK_SERVICE", "NETWORK_LINK_SERVICE_ASSOCIATION", "NS_NOTIFICATION_TYPE", "NS_SUBSCRIPTION", "NS_INTEGRATION", "NS_TEMPLATE", "NS_NOTIFICATION", "PRIVATE_LINK_ATTACHMENT", "PRIVATE_LINK_ATTACHMENT_CONNECTION" ] }, "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." } } }
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).
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/).Click through to view the messages on your dedicated cluster.
Click on the See in Stream Lineage in the top right to see which consumer groups are consuming the audit logs topic.
Use Terraform to create a cluster link to the Audit Logs cluster¶
The Confluent Terraform provider can be used to create a cluster link to mirror audit logs data to a Dedicated Confluent Cloud cluster.
When doing this, make sure to reference the Audit Logs environment and Kafka cluster by the ID and not the display_name. To be specific, reference the Audit Logs like this:
data "confluent_kafka_cluster" "audit-cluster" {
id = "lkc-abc123"
environment {
id = "env-xyz456"
}
}
Here is a full example of a Terraform resource for a cluster link from the audit logs cluster:
terraform {
required_providers {
confluent = {
source = "confluentinc/confluent"
version = "<most recent version>"
}
}
}
provider "confluent" {}
data "confluent_service_account" "audit-account" {
id = "<ID of the Audit Logs service account>"
}
data "confluent_service_account" "app-provisioning-account" {
display_name = "tf_runner"
}
data "confluent_kafka_cluster" "audit-cluster" {
id = "<audit logs cluster ID>"
environment {
id = "<audit logs environment ID>"
}
}
data "confluent_environment" "env" {
display_name = "<environment of Confluent Cloud Dedicated cluster>"
}
data "confluent_kafka_cluster" "dedicated" {
display_name = "<name of Confluent Cloud Dedicated cluster>"
environment {
id = data.confluent_environment.env.id
}
}
resource "confluent_role_binding" "app-provisioning-account-cluster-admin" {
principal = "User:${data.confluent_service_account.app-provisioning-account.id}"
role_name = "CloudClusterAdmin"
crn_pattern = data.confluent_kafka_cluster.dedicated.rbac_crn
}
resource "confluent_api_key" "audit-account-audit-cluster-api-key" {
display_name = "audit-account-audit-cluster-api-key"
description = "Kafka API Key for the Audit Logs cluster"
owner {
id = data.confluent_service_account.audit-account.id
api_version = data.confluent_service_account.audit-account.api_version
kind = data.confluent_service_account.audit-account.kind
}
managed_resource {
id = data.confluent_kafka_cluster.audit-cluster.id
api_version = data.confluent_kafka_cluster.audit-cluster.api_version
kind = data.confluent_kafka_cluster.audit-cluster.kind
environment {
id = data.confluent_kafka_cluster.audit-cluster.environment.0.id
}
}
}
resource "confluent_api_key" "app-provisioning-kafka-cluster-api-key" {
display_name = "app-provisioning-kafka-cluster-api-key"
description = "Kafka API Key for the dedicated destination cluster"
owner {
id = data.confluent_service_account.app-provisioning-account.id
api_version = data.confluent_service_account.app-provisioning-account.api_version
kind = data.confluent_service_account.app-provisioning-account.kind
}
managed_resource {
id = data.confluent_kafka_cluster.dedicated.id
api_version = data.confluent_kafka_cluster.dedicated.api_version
kind = data.confluent_kafka_cluster.dedicated.kind
environment {
id = data.confluent_kafka_cluster.dedicated.environment.0.id
}
}
depends_on = [
confluent_role_binding.app-provisioning-account-cluster-admin
]
}
resource "confluent_cluster_link" "audit-logs" {
link_name = "audit-link"
source_kafka_cluster {
id = data.confluent_kafka_cluster.audit-cluster.id
bootstrap_endpoint = data.confluent_kafka_cluster.audit-cluster.bootstrap_endpoint
credentials {
key = confluent_api_key.audit-account-audit-cluster-api-key.id
secret = confluent_api_key.audit-account-audit-cluster-api-key.secret
}
}
destination_kafka_cluster {
id = data.confluent_kafka_cluster.dedicated.id
rest_endpoint = data.confluent_kafka_cluster.dedicated.rest_endpoint
credentials {
key = confluent_api_key.app-provisioning-kafka-cluster-api-key.id
secret = confluent_api_key.app-provisioning-kafka-cluster-api-key.secret
}
}
lifecycle {
prevent_destroy = true
}
}
resource "confluent_kafka_mirror_topic" "audit-log-mirror-topic" {
source_kafka_topic {
topic_name = "confluent-audit-log-events"
}
cluster_link {
link_name = confluent_cluster_link.audit-logs.link_name
}
kafka_cluster {
id = data.confluent_kafka_cluster.dedicated.id
rest_endpoint = data.confluent_kafka_cluster.dedicated.rest_endpoint
credentials {
key = confluent_api_key.app-provisioning-kafka-cluster-api-key.id
secret = confluent_api_key.app-provisioning-kafka-cluster-api-key.secret
}
}
depends_on = [
confluent_cluster_link.audit-logs
]
}
resource "confluent_kafka_mirror_topic" "connect-events-mirror-topic" {
source_kafka_topic {
topic_name = "confluent-connect-log-events"
}
cluster_link {
link_name = confluent_cluster_link.audit-logs.link_name
}
kafka_cluster {
id = data.confluent_kafka_cluster.dedicated.id
rest_endpoint = data.confluent_kafka_cluster.dedicated.rest_endpoint
credentials {
key = confluent_api_key.app-provisioning-kafka-cluster-api-key.id
secret = confluent_api_key.app-provisioning-kafka-cluster-api-key.secret
}
}
depends_on = [
confluent_cluster_link.audit-logs
]
}