CI/CD and automation

Applies to VAST Cluster 5.5.

This page shows how to manage VAST Event Broker objects from a pipeline: the Kafka view, its topics, and blob expansions that turn a topic's JSON events into table columns. It covers the VMS REST API, vastpy, the VAST Terraform provider, and what the VAST admin MCP server can and can't do. Every example was run on a VAST Cluster 5.5.0-SP6 lab cluster.

Application code can also create and configure topics with the Kafka Admin API; see manage topics. The tools on this page are the right choice when topics are infrastructure: declared in Git, reviewed, and applied by a pipeline. For the shared setup (API tokens, TLS, users, keys and Terraform state), see CI/CD and automation for VAST DataBase. To choose between the tools, see automation tools compared.

What each tool can do

Task	REST API	`vastpy`	Terraform (4.0.2)	VAST admin MCP server
Create a Kafka view	Yes, on a VIP pool from your platform team	Yes	`vastdata_view`	No: the create fails
Create, change and delete topics	Yes	Yes	`vastdata_topic`	No tools
Compacted topics, `max_message_bytes`	Yes	Yes	No arguments for them	No
Create a blob expansion	Yes, after its target table exists	Yes	`vastdata_blob_expansion`, after its target table exists	No tools
Create the target table	Yes (`/schemas/`, `/tables/`)	Yes	No resource	No
List Kafka views	Yes	Yes	Data source	`list_views_vast`

vastdata_kafka_broker in the Terraform provider registers an external (third-party) event broker with the cluster. It doesn't create or configure VAST Event Broker; use vastdata_view and vastdata_topic for that.

Before you start

Admin task: send to your platform team. Two things on this page need them:

A VIP pool for each Kafka view. A cluster administrator creates it, with the Protocols role. See Configuring VAST Event Broker in the KB and setup. A pool that another Kafka view already uses only accepts views that require mTLS, so a second view with unencrypted connections needs its own pool (below).
An API token for the automation identity. The examples use a tenant administrator's VMS API token. See Managing VMS API tokens and CI/CD and automation for VAST DataBase.

Call a versioned API path, /api/v8/ or /api/latest/ (they return the same objects on 5.5). The unversioned /api/views/ returns an older shape of the view without any of the kafka_* fields, so a script that reads or copies a view through it drops the Event Broker settings without an error. vastpy uses /api/latest/.

API documentation

The KB's VAST Cluster 5.5 REST API reference: Create View, List Kafka Topics, Create a Kafka Topic, Modify a Kafka Topic, Show a Kafka Topic, Delete a Kafka Topic, Create a Blob Expansion, Show a Blob Expansion and Add Columns to a Blob Expansion.
Your cluster's own copy at https://<VMS address>/docs/ (VMS REST API documentation). The Swagger UI at https://<VMS address>/api/ lists only a few endpoints, not these.

REST API

The examples use VMS, the VMS address, and VMS_TOKEN, a tenant admin's API token. -k skips TLS verification for a lab cluster with a self-signed certificate; in a pipeline, pass the cluster's CA with --cacert instead.

Create the Kafka view

A Kafka view has the KAFKA, S3 and DATABASE protocols, a bucket owner, an S3 Native view policy, and its VIP pool in kafka_vip_pools. Use the ids of the view policy and of the pool your platform team created. Here the first attempt names a pool that another Kafka view (view-6) already uses:

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X POST "https://$VMS/api/v8/views/" -d '{"path": "/my-events", "bucket": "my-events", "bucket_owner": "my-events-owner",
  "policy_id": 7, "protocols": ["KAFKA", "S3", "DATABASE"], "kafka_vip_pools": [2], "create_dir": true}'

{"detail":"Only MTLS authentication is supported because the Kafka VIP is used by other views: view-6","code":"bad_request"}
HTTP 400

With a pool of its own, the view is created. By default it accepts unencrypted connections with no authentication:

curl -sk -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X POST "https://$VMS/api/v8/views/" -d '{"path": "/my-events", "bucket": "my-events", "bucket_owner": "my-events-owner",
  "policy_id": 7, "protocols": ["KAFKA", "S3", "DATABASE"], "kafka_vip_pools": [3], "create_dir": true}' \
  | jq "{id, path, bucket, protocols, kafka_vip_pools, tenant_id, kafka_unencrypted_auth_mechanism, is_kafka_unencrypted_conn_allowed}"

{
  "id": 9,
  "path": "/my-events",
  "bucket": "my-events",
  "protocols": [
    "KAFKA",
    "S3",
    "DATABASE"
  ],
  "kafka_vip_pools": [
    3
  ],
  "tenant_id": 3,
  "kafka_unencrypted_auth_mechanism": "NONE",
  "is_kafka_unencrypted_conn_allowed": true
}

For TLS, SASL/PLAIN, mTLS and authorization settings, see security. Decide them before the view is created (setup).

Create topics

POST /topics/ takes the view's bucket as database_name. It returns 201 with an empty body:

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X POST "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders", "topic_partitions": 4, "retention_ms": 86400000}'


HTTP 201

A compacted topic takes cleanup_policy, min_compaction_lag_ms and max_message_bytes. The Terraform provider has no arguments for these, so create compacted topics through the REST API, vastpy or the Kafka Admin API:

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X POST "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "customers", "topic_partitions": 1,
  "cleanup_policy": "compact", "min_compaction_lag_ms": 60000, "max_message_bytes": 1048576}'


HTTP 201

What compaction does, and its limits, is in manage topics and topic configuration.

List and show topics

GET /topics/ is paginated and returns only names:

curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/?database_name=my-events" | jq .

{
  "count": 2,
  "next": null,
  "previous": null,
  "results": [
    {
      "database_name": "my-events",
      "name": "orders"
    },
    {
      "database_name": "my-events",
      "name": "customers"
    }
  ]
}

GET /topics/show/ returns a topic's settings. Most numbers come back as strings, so compare them as strings in a script:

curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/show/?database_name=my-events&name=customers" | jq .

{
  "database_name": "my-events",
  "name": "customers",
  "num_events": 0,
  "capacity": 0,
  "topic_partitions": 1,
  "cleanup_policy": "compact",
  "min_compaction_lag_ms": "60000",
  "max_message_bytes": "1048576",
  "message_timestamp_type": "CreateTime",
  "delete_retention_ms": "86400000",
  "message_timestamp_after_max_ms": "9223372036854775807",
  "message_timestamp_before_max_ms": "9223372036854775807",
  "retention_ms": "604800000"
}

Change a topic

PATCH /topics/ changes settings in place and returns 204. The partition count is fixed when the topic is created.

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X PATCH "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders", "retention_ms": 172800000}'


HTTP 204

curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/show/?database_name=my-events&name=orders" | jq .

{
  "database_name": "my-events",
  "name": "orders",
  "num_events": 0,
  "capacity": 0,
  "topic_partitions": 4,
  "message_timestamp_type": "CreateTime",
  "retention_ms": "172800000",
  "cleanup_policy": "delete",
  "delete_retention_ms": "86400000",
  "max_message_bytes": "1048588",
  "message_timestamp_after_max_ms": "9223372036854775807",
  "message_timestamp_before_max_ms": "9223372036854775807",
  "min_compaction_lag_ms": "0"
}

Smoke-test the topic

End a pipeline by producing to the topic and reading the record back. The script uses confluent-kafka, the supported client (setup), with the producer settings VAST Event Broker needs. It reads the record back from the partition it was written to, so it doesn't create a consumer group. Save it as kafka_smoke.py:

"""Produce one record to a topic and read it back (smoke test after a CI job).
usage: python kafka_smoke.py <bootstrap host:port> <topic>"""
import sys, time, uuid
from confluent_kafka import Producer, Consumer, TopicPartition

bootstrap, topic = sys.argv[1], sys.argv[2]
marker = f"ci-smoke-{uuid.uuid4().hex[:8]}"
p = Producer({"bootstrap.servers": bootstrap, "acks": "all",
              "enable.idempotence": False, "compression.type": "none"})
reports = []
p.produce(topic, key=b"smoke", value=marker.encode(), on_delivery=lambda err, msg: reports.append((err, msg)))
p.flush(30)
if not reports or reports[0][0] is not None:
    sys.exit(f"produce failed: {reports[0][0] if reports else 'no delivery report'}")
md = reports[0][1]
print(f"produced {marker} to {md.topic()} partition {md.partition()} offset {md.offset()}")
# Read the partition directly from the produced offset: no consumer group is created.
c = Consumer({"bootstrap.servers": bootstrap, "group.id": "ci-smoke", "enable.auto.commit": False})
c.assign([TopicPartition(topic, md.partition(), md.offset())])
deadline = time.time() + 15
while time.time() < deadline:
    rec = c.poll(1.0)
    if rec is None or rec.error():
        continue
    if rec.value().decode() == marker:
        print(f"consumed {rec.value().decode()} from partition {rec.partition()} offset {rec.offset()}")
        break
else:
    c.close()
    sys.exit("marker not consumed")
c.close()

python kafka_smoke.py <Kafka VIP>:9092 orders

produced ci-smoke-5fb6b031 to orders partition 2 offset 0
consumed ci-smoke-5fb6b031 from partition 2 offset 0

The script exits non-zero if the record doesn't come back, which fails the job.

vastpy

vastpy reads the VMS address and token from VMS_ADDRESS and VMS_TOKEN, and doesn't verify TLS unless you set VMS_CERT_FILE (details). The same examples use the view my-events from the REST section.

vastpy-cli prints tables, which is handy for a pipeline's log:

vastpy-cli get topics database_name=my-events

database_name |name           
--------------+---------------+
my-events     |orders         
my-events     |customers

Keep topics in Git

Declare the topics in a YAML file, and let a script create the missing ones and correct drifted settings. It's safe to run on every commit. topics.yaml:

database_name: my-events
topics:
  - name: orders
    topic_partitions: 4
    retention_ms: 86400000
  - name: payments
    topic_partitions: 2
    retention_ms: 259200000

ensure_topics.py:

"""Ensure the Event Broker topics listed in topics.yaml exist (create missing, fix drifted settings).
Env: VMS_ADDRESS, VMS_TOKEN (tenant-admin API token)."""
import os
import yaml
from vastpy import VASTClient

client = VASTClient(address=os.environ["VMS_ADDRESS"], token=os.environ["VMS_TOKEN"])
spec = yaml.safe_load(open("topics.yaml"))
db = spec["database_name"]

existing = {t["name"] for t in client.topics.get(database_name=db)["results"]}
for want in spec["topics"]:
    name = want["name"]
    if name not in existing:
        client.topics.post(database_name=db, **want)
        print(f"created {db}/{name}")
        continue
    have = client.topics.show.get(database_name=db, name=name)
    changes = {k: v for k, v in want.items()
               if k not in ("name", "topic_partitions") and str(have.get(k)) != str(v)}
    if changes:
        client.topics.patch(database_name=db, name=name, **changes)
        print(f"updated {db}/{name}: {changes}")
    else:
        print(f"ok {db}/{name}")

The script never changes topic_partitions (it's fixed at creation), and compares values as strings because topics/show returns them as strings. It reads one page of GET /topics/; for a view with many topics, follow next. It doesn't delete topics that are missing from the file. That's deliberate: deleting a topic deletes its events.

The first run creates payments and puts back the retention of orders, which the REST section changed to two days:

python ensure_topics.py

updated my-events/orders: {'retention_ms': 86400000}
created my-events/payments

The second run changes nothing:

python ensure_topics.py

ok my-events/orders
ok my-events/payments

vastpy-cli get topics/show database_name=my-events name=payments

property                        |value               
--------------------------------+--------------------+
database_name                   |my-events           
name                            |payments            
num_events                      |0                   
capacity                        |0                   
topic_partitions                |2                   
retention_ms                    |259200000           
message_timestamp_type          |CreateTime          
cleanup_policy                  |delete              
delete_retention_ms             |86400000            
max_message_bytes               |1048588             
message_timestamp_after_max_ms  |9223372036854775807 
message_timestamp_before_max_ms |9223372036854775807 
min_compaction_lag_ms           |0

To keep compacted topics in the same file, add their cleanup_policy and related settings; the script passes every field through to the API.

Create a blob expansion

A blob expansion parses each event's JSON value into columns of a target table. The target table must exist before you create the expansion, with the same columns. Otherwise the create fails with 404:

vastpy.RESTFailure: response for POST api/latest/blobexpansions with params=None failed with error 404 and message {"detail":"Creating blob_expansion failed. [req=0x5011000588b5]: failed to get target table for blob expansion in @s3_bucket_name=my-events"}

So the script creates the schema and table through VMS first. blob_expansion.py:

"""Expand the JSON value of topic orders into a table (blob expansion), then show it.
The target table must exist first, so create its schema and table through VMS too.
Env: VMS_ADDRESS, VMS_TOKEN (tenant-admin API token)."""
import json, os
from vastpy import VASTClient

client = VASTClient(address=os.environ["VMS_ADDRESS"], token=os.environ["VMS_TOKEN"])
DB, TOPIC, SCHEMA, TARGET = "my-events", "orders", "expanded", "orders_expanded"
columns = [
    {"name": "order_id", "field": {"column_type": "int64"}},
    {"name": "sku", "field": {"column_type": "string"}},
]
client.schemas.post(database_name=DB, name=SCHEMA)
client.tables.post(database_name=DB, schema_name=SCHEMA, name=TARGET, arrow_schema=columns)
client.blobexpansions.post(
    database_name=DB,
    table_name=TOPIC,
    target_table_schema=SCHEMA,
    target_table_name=TARGET,
    expansion_format="json",
    arrow_schema=columns,
)
show = client.blobexpansions.show.get(database_name=DB, table_name=TOPIC)
print(json.dumps(show, indent=2))

python blob_expansion.py

{
  "target_table_name": "/<tenant>/my-events/expanded/orders_expanded",
  "expansion_format": "json",
  "copy_source_column": false,
  "source_column_name": "value",
  "columns": [
    {
      "database_name": "my-events",
      "schema_name": "kafka_topics",
      "table_name": "orders",
      "name": "order_id",
      "field": {
        "column_type": "int64"
      },
      "raw_field": "int64"
    },
    {
      "database_name": "my-events",
      "schema_name": "kafka_topics",
      "table_name": "orders",
      "name": "sku",
      "field": {
        "column_type": "string"
      },
      "raw_field": "string"
    }
  ],
  "add_missing_values_output": true,
  "add_excessive_values_output": true
}

The expansion is addressed by the topic: database_name is the view's bucket and table_name is the topic, which is a table in the kafka_topics schema.

Add a field later with blobexpansions/add_columns. The command prints only a blank line:

vastpy-cli patch blobexpansions/add_columns database_name=my-events table_name=orders \
  arrow_schema='[{"name": "qty", "field": {"column_type": "int32"}}]'

vastpy-cli get blobexpansions/show database_name=my-events table_name=orders --json | jq -c ".columns[] | {name, raw_field}"

{"name":"order_id","raw_field":"int64"}
{"name":"sku","raw_field":"string"}
{"name":"qty","raw_field":"int32"}

The other settings (copy_source_column, add_missing_values_output and so on), what happens to events from before the expansion, and how to remove one are in expand JSON events into columns.

The application that reads the target table with the VAST DataBase SDK needs an identity policy with the s3:Tabular* actions on the view's bucket. Owning the bucket isn't enough: without the policy, opening the table fails with vastdb.errors.Forbidden ('code': 'AccessDenied', "Permission check failed"). The Terraform example below creates such a user and policy.

Clean up

Remove what the REST and vastpy examples created, in this order: the blob expansion, its target table and schema, the topics, the kafka_topics schema, and the view.

vastpy-cli delete blobexpansions database_name=my-events table_name=orders && echo "blob expansion removed"


blob expansion removed

Tables and schemas are deleted with a JSON body:

curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X DELETE "https://$VMS/api/v8/tables/" -d '{"database_name": "my-events", "schema_name": "expanded", "name": "orders_expanded"}'
curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X DELETE "https://$VMS/api/v8/schemas/" -d '{"database_name": "my-events", "name": "expanded"}'

HTTP 204
HTTP 204

So are topics. Repeat for each topic:

curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X DELETE "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders"}'

HTTP 204

A Kafka view can't be deleted while it holds any schema, and the cluster creates one, kafka_topics, in every Kafka view. Deleting all the topics isn't enough:

curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -X DELETE "https://$VMS/api/v8/views/9/"

{"detail":"Deleting a view with created schemas or tables is not allowed","code":"bad_request"}HTTP 400

Drop the kafka_topics schema (and any target schemas of blob expansions), then delete the view:

curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \
  -X DELETE "https://$VMS/api/v8/schemas/" -d '{"database_name": "my-events", "name": "kafka_topics"}'
curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -X DELETE "https://$VMS/api/v8/views/9/"

HTTP 204
HTTP 204

The VIP pool can't be deleted while a Kafka view uses it (409, "Unable to delete … because it is being used by the following Kafka views"). Removing it is your platform team's task.

Terraform

The provider (version 4.0.2) has three resources for Event Broker: vastdata_view (the Kafka view), vastdata_topic and vastdata_blob_expansion. It has none for schemas or tables, so a blob expansion's target table is created outside Terraform.

vastdata_topic has topic_partitions, retention_ms and the message_timestamp_* settings, but no cleanup_policy, compaction settings or max_message_bytes. Create compacted topics with the REST API or vastpy (above), and keep them out of Terraform so the two don't fight.

This configuration creates a Kafka view on a VIP pool your platform team made, its topics from a map with for_each, an application user with an identity policy and a key (for reading topic and target tables with the SDK), and, when expand_orders is true, a blob expansion of orders:

# VAST Event Broker: a Kafka view on a VIP pool the platform team provides, its topics,
# an application user + identity policy + key (for SDK reads of topic tables), and a blob
# expansion of one topic into a table.
# Exercised against VAST Cluster 5.5.0-SP6 with provider 4.0.2 as a tenant admin (API token).
terraform {
  required_version = ">= 1.5"
  required_providers {
    vastdata = {
      source  = "vast-data/vastdata"
      version = "4.0.2"
    }
  }
}

# In CI: TF_VAR_vast_host, VASTDATA_API_TOKEN. Provider 4.0.2 needs host in the configuration.
variable "vast_host" {
  type = string
}

provider "vastdata" {
  host            = var.vast_host
  skip_ssl_verify = true # lab cluster with a self-signed certificate
}

variable "tenant_name" {
  description = "Tenant that holds the Kafka view"
  type        = string
}

variable "kafka_vip_pool" {
  description = "Name of a PROTOCOLS VIP pool for this view (from the platform team). A pool already used by another Kafka view only accepts mTLS views."
  type        = string
  default     = "events-ci-pool"
}

variable "bucket" {
  type    = string
  default = "events-ci"
}

variable "topics" {
  type = map(object({
    partitions   = number
    retention_ms = number
  }))
  default = {
    "orders"   = { partitions = 4, retention_ms = 86400000 }
    "payments" = { partitions = 2, retention_ms = 259200000 }
  }
}

variable "pgp_public_key" {
  type = string
}

variable "expand_orders" {
  description = "Create the blob expansion. The target table must exist first (no Terraform resource for tables)."
  type        = bool
  default     = false
}

data "vastdata_tenant" "this" {
  name = var.tenant_name
}

data "vastdata_vip_pool" "kafka" {
  name = var.kafka_vip_pool
}

resource "vastdata_s3_policy" "events" {
  name      = "${var.bucket}-tabular"
  tenant_id = data.vastdata_tenant.this.id
  enabled   = true
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect   = "Allow"
      Action   = ["s3:ListBucket", "s3:Tabular*"]
      Resource = [var.bucket, "${var.bucket}/*"]
    }]
  })
}

resource "vastdata_user" "app" {
  name              = "${var.bucket}-app"
  uid               = 8205
  local_provider_id = data.vastdata_tenant.this.local_provider_id
  s3_policies_ids   = [vastdata_s3_policy.events.id]
}

resource "vastdata_user_key" "app" {
  user_id        = vastdata_user.app.id
  tenant_id      = data.vastdata_tenant.this.id
  pgp_public_key = var.pgp_public_key
}

resource "vastdata_view_policy" "events" {
  name      = "${var.bucket}-policy"
  tenant_id = data.vastdata_tenant.this.id
  flavor    = "S3_NATIVE"
}

resource "vastdata_view" "events" {
  path                              = "/${var.bucket}"
  bucket                            = var.bucket
  bucket_owner                      = vastdata_user.app.name
  protocols                         = ["KAFKA", "S3", "DATABASE"]
  policy_id                         = vastdata_view_policy.events.id
  tenant_id                         = data.vastdata_tenant.this.id
  create_dir                        = true
  kafka_vip_pools                   = [data.vastdata_vip_pool.kafka.id]
  is_kafka_unencrypted_conn_allowed = true
  kafka_unencrypted_auth_mechanism  = "NONE"
}

resource "vastdata_topic" "this" {
  for_each         = var.topics
  database_name    = vastdata_view.events.bucket
  name             = each.key
  topic_partitions = each.value.partitions
  retention_ms     = each.value.retention_ms
  tenant_id        = data.vastdata_tenant.this.id
}

resource "vastdata_blob_expansion" "orders" {
  count               = var.expand_orders ? 1 : 0
  database_name       = vastdata_view.events.bucket
  table_name          = vastdata_topic.this["orders"].name
  target_table_schema = "expanded"
  target_table_name   = "orders"
  expansion_format    = "json"
  arrow_schema = [
    { name = "order_id", field = { column_type = "int64" } },
    { name = "sku", field = { column_type = "string" } },
  ]
  tenant_id = data.vastdata_tenant.this.id
}

output "bootstrap_servers" {
  value = [for ip in data.vastdata_vip_pool.kafka.ip_ranges : "${ip[0]}:9092"]
}

output "access_key" {
  value = vastdata_user_key.app.access_key
}

output "encrypted_secret_key" {
  value = vastdata_user_key.app.encrypted_secret_key
}

The view accepts unencrypted connections without authentication, which suits a lab. For TLS, SASL/PLAIN or mTLS, set the view's kafka_* arguments to match what your platform team agreed; see security. bootstrap_servers takes the first address of each range in the pool; list three addresses if you have them (setup).

Apply

The provider reads the token from VASTDATA_API_TOKEN; pass the VMS address as TF_VAR_vast_host and the tenant as TF_VAR_tenant_name. The user key's secret is encrypted with an ASCII-armored PGP public key.

terraform init -no-color | grep -E "Installed|successfully"

- Installed vast-data/vastdata v4.0.2 (self-signed, key ID FADBA165459C78A3)
Terraform has been successfully initialized!

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" terraform apply -no-color -auto-approve \
  | grep -E "Creat|Apply|Error|error|^Outputs|^[a-z_]+ = |:9092"

      + "<Kafka VIP>:9092",
vastdata_s3_policy.events: Creating...
vastdata_view_policy.events: Creating...
vastdata_s3_policy.events: Creation complete after 1s [name=events-ci-tabular]
vastdata_user.app: Creating...
vastdata_user.app: Creation complete after 2s [name=events-ci-app]
vastdata_user_key.app: Creating...
vastdata_user_key.app: Creation complete after 1s
vastdata_view_policy.events: Creation complete after 5s [name=events-ci-policy]
vastdata_view.events: Creating...
vastdata_view.events: Creation complete after 10s [name=view-19]
vastdata_topic.this["orders"]: Creating...
vastdata_topic.this["payments"]: Creating...
vastdata_topic.this["orders"]: Creation complete after 3s [name=orders]
vastdata_topic.this["payments"]: Creation complete after 5s [name=payments]
Apply complete! Resources: 7 added, 0 changed, 0 destroyed.
Outputs:
access_key = "<access key>"
bootstrap_servers = [
  "<Kafka VIP>:9092",
encrypted_secret_key = <<EOT

Smoke-test a topic through the output's bootstrap address, with kafka_smoke.py from the REST section:

python kafka_smoke.py $(terraform output -json bootstrap_servers | jq -r ".[0]") payments

produced ci-smoke-5fabd3e7 to payments partition 0 offset 0
consumed ci-smoke-5fabd3e7 from partition 0 offset 0

Drift

The provider detects changes made outside Terraform and puts them back on the next apply. Here someone shortens the retention of orders with vastpy-cli:

vastpy-cli patch topics database_name=events-ci name=orders retention_ms=3600000 && echo patched


patched

The next plan shows the change it will revert:

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" terraform plan -no-color \
  | grep -vE "Refreshing|Reading|Read complete|^\s*$"

Terraform used the selected providers to generate the following execution
plan. Resource actions are indicated with the following symbols:
  ~ update in-place
Terraform will perform the following actions:
  # vastdata_topic.this["orders"] will be updated in-place
  ~ resource "vastdata_topic" "this" {
        name                            = "orders"
      ~ retention_ms                    = 3600000 -> 86400000
        # (8 unchanged attributes hidden)
    }
Plan: 0 to add, 1 to change, 0 to destroy.

So if Terraform owns a topic, change it in Terraform, not with the Kafka Admin API or the other tools: anything else is undone by the next apply. A scheduled terraform plan -detailed-exitcode job is a simple drift alarm.

Add the blob expansion

vastdata_blob_expansion fails until its target table exists. The same apply still corrects the drifted topic:

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \
  terraform apply -no-color -auto-approve | grep -E "Creat|Modif|Apply|Error|error|│"


Error: error creating "blob_expansion" resource.

  with vastdata_blob_expansion.orders[0],
  on main.tf line 125, in resource "vastdata_blob_expansion" "orders":
 125: resource "vastdata_blob_expansion" "orders" {

POST request to https://<VMS address>:443/api/latest/blobexpansions/ returned
status code 404 — response body: {
  "detail": "Creating blob_expansion failed. [req=0x50110006c93e]: failed to get target table for blob expansion in @s3_bucket_name=events-ci"
}
vastdata_topic.this["orders"]: Modifying... [name=orders]
vastdata_topic.this["orders"]: Modifications complete after 0s [name=orders]
vastdata_blob_expansion.orders[0]: Creating...

Create the schema and the target table with the same columns, here with vastpy-cli (or the SDK, or the REST API):

vastpy-cli post schemas database_name=events-ci name=expanded && \
  vastpy-cli post tables database_name=events-ci schema_name=expanded name=orders \
    arrow_schema='[{"name": "order_id", "field": {"column_type": "int64"}}, {"name": "sku", "field": {"column_type": "string"}}]' && \
  echo "target table created"


target table created

Then apply again:

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \
  terraform apply -no-color -auto-approve | grep -E "Creat|Modif|Apply|Error|error|│"

vastdata_blob_expansion.orders[0]: Creating...
vastdata_blob_expansion.orders[0]: Creation complete after 1s
Apply complete! Resources: 1 added, 0 changed, 0 destroyed.

In a pipeline, put the table step before terraform apply, and make it re-runnable (check for the table first, or use the SDK's create_table(..., fail_if_exists=False)).

To check the expansion, produce two JSON orders and read the target table with the SDK, using the application key from the Terraform outputs. verify_expansion.py:

"""Produce two JSON orders to orders, then read the expanded rows with the VAST DataBase SDK.
Env: S3_AK, S3_SK (the application user's key pair). Args: <kafka bootstrap> <db endpoint>"""
import json, os, sys
from confluent_kafka import Producer
import vastdb

bootstrap, endpoint = sys.argv[1], sys.argv[2]
p = Producer({"bootstrap.servers": bootstrap, "acks": "all",
              "enable.idempotence": False, "compression.type": "none"})
for order in ({"order_id": 1001, "sku": "A-1", "qty": 2}, {"order_id": 1002, "sku": "B-7", "qty": 1}):
    p.produce("orders", value=json.dumps(order).encode())
if p.flush(30):
    sys.exit("orders not delivered")

session = vastdb.connect(endpoint=endpoint, access=os.environ["S3_AK"], secret=os.environ["S3_SK"])
with session.transaction() as tx:
    t = tx.bucket("events-ci").schema("expanded").table("orders")
    print([f.name for f in t.columns()])
    rows = t.select(columns=["order_id", "sku", "qty"] if "qty" in [f.name for f in t.columns()] else ["order_id", "sku"]).read_all()
    print(rows.to_pandas().to_string(index=False))

S3_AK=$(terraform output -raw access_key) S3_SK=$(terraform output -raw encrypted_secret_key | gpg --batch --quiet --decrypt) \
  python verify_expansion.py <Kafka VIP>:9092 http://<Kafka VIP>:9090

['order_id', 'sku', 'blob_expansion_key', 'blob_expansion_timestamp_broker', 'blob_expansion_timestamp_producer', 'blob_expansion_headers', 'blob_expansion_error_output', 'blob_expansion_mismatches_output', 'blob_expansion_missing_values_output', 'blob_expansion_excessive_values_output', 'blob_expansion_kafka_partition', 'blob_expansion_external_offset']
 order_id sku
     1001 A-1
     1002 B-7

The cluster added the blob_expansion_* metadata columns to the target table; they're described in expand JSON events into columns. qty isn't a column because this expansion doesn't declare it. A plan afterwards reports no changes.

Destroy

terraform destroy fails on the view while it holds schemas: the target schema you created, and kafka_topics, which the cluster creates in every Kafka view. Resources ordered before the view (the key, the blob expansion and the topics) are already gone by then:

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \
  terraform destroy -no-color -auto-approve | grep -E "Destroy|destroy|Error|error|│|detail"


Error: Delete[view]

DELETE request to https://<VMS address>:443/api/latest/views/19/ returned
status code 400 — response body: {
  "detail": "Deleting a view with created schemas or tables is not allowed",
  "code": "bad_request"
}
  - destroy
  # vastdata_blob_expansion.orders[0] will be destroyed
  # vastdata_s3_policy.events will be destroyed
  # vastdata_topic.this["orders"] will be destroyed
  # vastdata_topic.this["payments"] will be destroyed
  # vastdata_user.app will be destroyed
  # vastdata_user_key.app will be destroyed
  # vastdata_view.events will be destroyed
  # vastdata_view_policy.events will be destroyed
Plan: 0 to add, 0 to change, 8 to destroy.
vastdata_user_key.app: Destroying...
vastdata_blob_expansion.orders[0]: Destroying...
vastdata_topic.this["payments"]: Destroying... [name=payments]
vastdata_topic.this["orders"]: Destroying... [name=orders]
vastdata_view.events: Destroying... [name=view-19]

Drop the target table and both schemas, then run destroy again:

vastpy-cli delete tables database_name=events-ci schema_name=expanded name=orders && \
  vastpy-cli delete schemas database_name=events-ci name=expanded && \
  vastpy-cli delete schemas database_name=events-ci name=kafka_topics && echo "schemas dropped"


schemas dropped

VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \
  terraform destroy -no-color -auto-approve | grep -E "Destroy|Destruction|Error|error|│|detail"

vastdata_view.events: Destroying... [name=view-19]
vastdata_view.events: Destruction complete after 5s
vastdata_user.app: Destroying... [name=events-ci-app]
vastdata_view_policy.events: Destroying... [name=events-ci-policy]
vastdata_view_policy.events: Destruction complete after 1s
vastdata_user.app: Destruction complete after 4s
vastdata_s3_policy.events: Destroying... [name=events-ci-tabular]
vastdata_s3_policy.events: Destruction complete after 0s
Destroy complete! Resources: 4 destroyed.

Destroying topics deletes their events. Guard destroy jobs with a manual approval, and consider lifecycle { prevent_destroy = true } on topics that hold data you need.

VAST admin MCP server

The VAST admin MCP server (vast-admin-mcp) has no tools for topics, blob expansions or brokers: none of its tools mentions them. It can list views, and so show which ones are Kafka views (protocols contains KAFKA), with list_views_vast. This output was filtered on the client to the Kafka views and three of their fields:

call list_views_vast {"path": "/*"}
isError: False
[
 {
  "path": "/events-ci",
  "bucket": "events-ci",
  "protocols": "S3,DATABASE,KAFKA"
 },
 {
  "path": "/my-events",
  "bucket": "my-events",
  "protocols": "S3,DATABASE,KAFKA"
 }
]

It can't create a Kafka view: its create_view_vast tool has no VIP pool argument, and the cluster refuses the view without one.

call create_view_vast {"cluster": "<cluster>", "tenant": "<tenant>", "path": "/my-events-mcp", "protocols": "KAFKA,S3,DATABASE", "bucket": "my-events-mcp", "bucket_owner": "my-app", "policy": "my-db-policy"}
isError: True
Error calling tool 'create_view_vast': response for POST api/latest/views with params=None failed with error 400 and message {"detail":"VIP pools must be provided for a view with the KAFKA protocol","code":"bad_request"}

For an assistant that works with topics, give it a Kafka client and the Kafka Admin API instead. Setup and the server's other tools are in CI/CD and automation for VAST DataBase.

A CI job

This GitLab CI job applies topics.yaml with ensure_topics.py on every push to the default branch, then smoke-tests one topic. It was not run end to end on a CI runner. Each step it runs was run by hand on a 5.5.0-SP6 lab cluster, as shown above.

# Not run end to end on GitLab. The steps match the vastpy and smoke-test sections of this page.
#
# CI/CD variables (Settings > CI/CD > Variables):
#   VMS_ADDRESS        VMS address
#   VMS_TOKEN          tenant-admin API token (masked, protected)
#   KAFKA_BOOTSTRAP    <Kafka VIP>:9092 of the view in topics.yaml
# The runner must reach VMS on port 443 and every VIP in the Kafka view's pool.
stages: [topics, verify]

ensure-topics:
  stage: topics
  image: python:3.12-slim
  resource_group: event-broker-topics   # one run at a time
  rules:
    - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH
      changes: [topics.yaml]
  before_script:
    - pip install vastpy==0.3.22 pyyaml
  script:
    - python ensure_topics.py

smoke-test:
  stage: verify
  image: python:3.12-slim
  needs: [ensure-topics]
  rules:
    - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH
      changes: [topics.yaml]
  before_script:
    - pip install confluent-kafka==2.8.0
  script:
    - python kafka_smoke.py "$KAFKA_BOOTSTRAP" orders

For a Terraform-based job, run terraform plan on merge requests and terraform apply on the default branch, with VASTDATA_API_TOKEN as a masked variable and remote state; CI/CD and automation for VAST DataBase covers state and secrets.