Publishing Events to VAST Event Broker

VAST Event Broker collects events occurring to elements in VAST Element Store into Kafka event topics. VAST Event Broker is based on the VAST Database which allows consumers to query event topics via the various database APIs.

VAST Event Broker utilizes VAST proprietary implementation of the Kafka protocol, as detailed in Kafka Protocol Support.

The workflow is as follows:

The VAST implementation of the Kafka protocol supports a basic subset of the Kafka APIs to allow clients to publish and consume events from the VAST Event Broker.

VAST Event Broker supports:

The following Kafka capabilities are not supported:

The following limitations apply:

VAST Cluster supports Confluent Kafka Python client 2.4 - 2.8.

Complete these steps to configure VAST Event Broker:

Create a virtual IP pool to be used to access the event topics using the Kafka protocol. The pool must have the PROTOCOLS role assigned.Managing Virtual IP Pools

If the VAST Event Broker view will be associated with a view policy that includes virtual IP pools, the pool specified as the Kafka pool must be one of the view policy pools.

Note

If you are going to use mTLS authentication for the VAST Event Broker, the virtual IP pool can be shared with VAST Event Broker views on other tenants, since determination of the correct tenant for each client request will be done based on the information in the CA certificate.

Configure a user on the authentication provider attached to the tenant where VAST Event Broker will run so that the user has S3 bucket creation permissions and S3 access keys.

This user will be assigned as the bucket owner when creating a view for VAST Event Broker.

VAST Event Broker runs on a view that has Kafka protocol enabled.

Note

View nesting is not supported.

Follow the steps in Creating Views to create a VAST Event Broker view as follows:

When creating a VAST Event Broker view in VAST CLI using the view create command, specify --protocols KAFKA to enable the Kafka protocol and --kafka-vip-pools <pool ID> to set the virtual IP pool. For example, to create a VAST Event Broker view accessible via virtual IP pool 3:

view create --tenant-id 1 --path </dir> --create-dir --policy-id 2 --protocols KAFKA,DATABASE,S3 --bucket mybucket --bucket-owner myadmin --kafka-vip-pools 3

(Optional) Configure authentication of users accessing the Kafka-enabled view.

Users accessing a Kafka-enabled view can be authenticated against the VAST provider associated with the tenant using the PLAIN Simple Authentication and Security Layer (SASL) mechanism. SASL/PLAIN authentication is available on both encrypted (TLS) and non-encrypted connections.

To configure user authentication:

If you are going to use SASL/PLAIN authentication with TLS, ensure that the VAST cluster has been configured with a TLS certificate intended for use with Kafka-enabled views. To check or upload a Kafka TLS certificate and a key:

The following limitations apply:

(Optional) Configure mutual TLS (mTLS) authentication between VAST Event Broker and event consumers, which allows the server to authenticate clients connecting to a Kafka-enabled view on the VAST cluster.

In multi-tenant environments, you can configure the VAST cluster to distinguish between mTLS certificates associated with particulars tenants, and thus to direct the IO to the intended tenant based on the mTLS certificate (in addition to using the client IP or destination virtual IP for that purpose). This improves security while enforcing tenant isolation and tenant-level authorization for NFS clients.

If mTLS authentication is enabled, connections without a proper CA certificate are rejected.

The CA certificate can be tenant-specific or global (cluster-wide, accepted by all tenants).

Note

With VAST Event Broker mTLS in place, VAST Event Broker views hosted on multiple tenants can use the same virtual IP pool, since determination of the correct tenant for each client request will be done based on the information in the CA certificate.

mTLS authentication for Kafka can be configured as follows:

After adding an mTLS certificate, you need to enable mTLS authentication for the VAST Event Broker view.

Cluster-wide configuration of mTLS authentication for Kafka enables certificate-based tenant identification.

The cluster-level CA certificate for mTLS authentication certificate is used for all tenants.

Warning

The cluster-wide configuration overrides any tenant-level settings. Applying the customer-wide configuration may cause temporary disconnection of tenant clients.

Complete these steps:

A tenant-level mTLS certificate must be unique across all tenants.

If the tenant is configured with a CA certificate and the client sends a certificate that is not signed by this CA (or a trust chain that does not provide for the signing), access is denied.

To add an mTLS certificate for a tenant:

To enable mTLS authentication for a VAST Event Broker view:

(Optional) Configure authorization of users accessing the Kafka-enabled view.

Only explicitly allowed operations can be performed.

You create, view, modify and delete event topics similar to VAST Database tables. When you create a VAST Event Broker view, VAST Cluster automatically creates an associated VAST database with a schema labelled Kafka-Compatible Broker Topics. Under this schema, you can create and manage topics where events will be recorded.

To create a topic in VAST Web UI:

To modify a topic in VAST Web UI:

Use the following VAST CLI commands to manage event topics:

You grant permissions for Kafka-related objects by listing operations that a user is allowed or prohibited to perform in the user's identity policy.

By default, if no ACL is found, the operation is allowed for any user. If you want to alter this behavior, contact VAST Support.

The following Kafka operations are supported:

For example, the following identity policy allows the user to read and write to the topic, but prohibits topic creation and deletion:

"Version": "2012-10-17",
"Statement": [
  {
    "Sid": "Aaa__gggg",
    "Action": "KafkaRead",
    "Effect": "Allow",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__ssss",
    "Action": "KafkaWrite",
    "Effect": "Allow",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__dddd",
    "Action": "KafkaCreate",
    "Effect": "Deny",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__cccc",
    "Action": "KafkaDelete",
    "Effect": "Deny",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
]

Prior to using Kafka APIs, ensure that the prerequisites described in Configuring VAST Event Broker are met.

The following examples use a Kafka client that can be installed with:  pip install kafka-python.

In the examples, you need to supply a list of Kafka bootstrap server IPs. These are the IPs that a Kafka client uses to connect to the cluster. It is recommended to specify three of the virtual IPs from the virtual IP pool assigned to the VAST Event Broker view.

from kafka.admin import KafkaAdminClient, NewTopic
BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
topic_name = #
# Kafka Admin Configuration
admin_client = KafkaAdminClient(
    bootstrap_servers=[BOOSTRAP_SERVERS],
    client_id='python-admin-client'
)
topic_name = topic_name
num_partitions = 2
replication_factor = 1
topic = NewTopic(
    name=topic_name,
    num_partitions=num_partitions,
    replication_factor=replication_factor
)
try:
    # Create the topic
    admin_client.create_topics(new_topics=[topic], validate_only=False)
    print(f"Topic '{topic_name}' created successfully with {num_partitions} partitions and replication factor {replication_factor}.")
except Exception as e:
    print(f"Error creating topic: {e}")
finally:
    admin_client.close()

BOOTSTRAP_SERVER = "<comma-separated list of IPs>"
from kafka.admin import KafkaAdminClient
# Define the Kafka broker (adjust the host/port as needed)
# Create an admin client
admin_client = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVER)
# Fetch the list of topics
topics = admin_client.list_topics()
# Print the list of topics
print("Kafka Topics:", topics)
# Close the admin client
admin_client.close()

from kafka import KafkaProducer
import json
BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
producer = KafkaProducer(bootstrap_servers=BOOSTRAP_SERVERS)
for i in range (100,2500):
    key = str(i)
    message = {"name": "vast", "id": i}
    producer.send(topic_name, key=key.encode("utf-8") , value=json.dumps(message).encode('utf-8'))
producer.flush()
print("Message sent to Kafka topic")

BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
from kafka import KafkaConsumer
consumer = KafkaConsumer(
    topic_name,
    bootstrap_servers=[BOOSTRAP_SERVERS],
    auto_offset_reset='earliest',
    group_id='your-consumer-group-name'
)
for message in consumer:
    print(f"Received message: {message.value.decode('utf-8')}")

from kafka.admin import KafkaAdminClient, NewTopic
# Define the Kafka broker
BOOTSTRAP_SERVER = "<comma-separated list of IPs>"
TOPIC_TO_DELETE = "my_topic"  # Change this to the topic you want to delete
# Create an admin client
admin_client = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVER)
# Delete the topic
try:
    admin_client.delete_topics([TOPIC_TO_DELETE])
    print(f"Topic '{TOPIC_TO_DELETE}' deleted successfully.")
except Exception as e:
    print(f"Failed to delete topic '{TOPIC_TO_DELETE}': {e}")
# Close the admin client
admin_client.close()

VAST Event Broker topics can be queried through the VAST Database API, and also using VAST Database connectors (e.g. the connectors for Trino and Spark).

The following Trino example shows the schema (kafka_topics) of the VAST Database used to store event topics, among other schemas on the VAST cluster:

trino> show schemas in vast;
                     Schema                      
-------------------------------------------------
 information_schema                              
 kafka2/kafka_topics                             
 system_schema                                   
 vast-audit-log-bucket/vast_audit_log_schema     
 vast-big-catalog-bucket/vast_big_catalog_schema 
(5 rows)

The following Trino example shows topics within the kafka_topics schema:

trino> show tables in vast."kafka2/kafka_topics";
         Table          
------------------------
 docker-connect-configs 
 docker-connect-offsets 
 docker-connect-status  
 my-new-topic           
 myfirsttopic           
 newtopic               
 streaming-topic                     
(7 rows)

VAST Cluster provides analytic reports that show latency and bandwidth for produced and fetched events and messages per VAST Event Broker view and per CNode.

To access Kafka-related analytics in VAST Web UI, go to Analytics -> Analytics -> Predefined Analytics tab, set the View or CNode filter and select one of the reports containing the word Kafka.

The following data protection capabilities are supported for VAST Event Broker topics:

The following rules and limitations apply:

Information in Kafka event messages can be parsed and saved in a VAST database table that can then be queried using any of the query methods available for VAST database tables.

You configure parsing by creating or selecting a VAST database table that will store parsed Kafka event data, and then selecting the specific fields to be parsed from the Kafka events. Columns corresponding to the these fields are added to the table (if not already present), and all subsequent events are parsed for these fields.

You can view statistics for parsed Kafka events using custom analytics in the Analytics page of the VAST Web UI.

In order to configure or modify event parsing configurations in the VAST Database, these S3 permissions are required:

These are described in Managing Permissions for Accessing VAST Tabular Databases

You can disable parsing of Kafka events for a Kafka topic. Events that have already been parsed and saved in the target table will remain, but new events will not be parsed.

Use these commands to configure and manage Kafka event parsing using the VAST CLI

This example illustrates how parsing can be configured for a Kafka topic, using the VAST Python SDK.

import pyarrow as pa
import vastdb  # pip install vastdb  (requires Python 3.10)
from vastdb.table import BlobExpansionConfig, ExpansionFormat

# Connection info
kafka_bucket  = "my_kafka_bucket"
kafka_schema  = "kafka_topics"      # the schema Kafka topics live under
topic_name    = "my_topic"          # source table = Kafka topic name
target_schema = "expanded_data"     # different schema, same bucket
target_table  = "my_expanded_table"

# NOTE: the source table (topic_name) and its Kafka topic must be created beforehand

# 1. Connect
session = vastdb.connect(access_key, secret_key, endpoint)

# 2. Create the target schema (different schema, same bucket)
with session.transaction() as tx:
    tx.bucket(kafka_bucket).create_schema(target_schema)

# 3. Define what the expanded blob looks like
expansion_schema = pa.schema([
    pa.field("event_type", pa.string()),
    pa.field("id",  pa.int64()),
    # ... any fields the JSON/blob contains
])

# 4. Create the target table in the target schema
with session.transaction() as tx:
    tx.bucket(kafka_bucket).schema(target_schema).create_table(
        target_table, expansion_schema, sorting_key=[]
    )

# 5. Create the blob expansion (links source blob column → target table)
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).create_blob_expansion(
        expansion_schema=expansion_schema,
        target_table_name=target_table,
        target_table_schema=target_schema,   # required: target is in a different schema
        source_column_name="value",
        config=BlobExpansionConfig(
            expansion_format=ExpansionFormat("json"),
            copy_source_column=False,
            flatten_path=False,
            flatten_delimiter="__",
        ),
    )

# 6. Add a column to the blob expansion
columns_to_add = pa.schema([
    pa.field("user_id", pa.string()),
])
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value")
    be.add_columns(columns_to_add=columns_to_add)

# 7. Remove a column from the blob expansion
columns_to_remove = pa.schema([
    pa.field("id", pa.int64()),
])
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value")
    be.drop_columns(columns_to_remove=columns_to_remove)

# 8. Drop the blob expansion entirely
with session.transaction() as tx:
    tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value").drop()