Introducing the Pulsar Admin Library for Go

Apache Pulsar is a highly scalable and reliable messaging system that is gaining popularity among developers. Pulsar provides a wide range of features and benefits that make it a popular choice for modern data streaming applications. However, managing a Pulsar cluster can be a complex task, which is why StreamNative has created the pulsar-admin-go library.

The pulsar-admin-go is a Go library that provides developers with a unified set of APIs to programmatically manage Pulsar clusters. It allows for easy automation of tasks and seamless integration of Pulsar management into your applications.

In this blog post, we'll take a closer look at the pulsar-admin-go library, its features, benefits, and advanced usage. We'll also provide step-by-step instructions on how to install and use the library.

I. Overview of pulsar-admin-go Library

The pulsar-admin-go library offers a range of useful management functionalities working with topics, partitions, and subscriptions.

• Topics: create, delete, get topic metadata, and list topics.
• Partitions: add and remove partitions from a topic.
• Subscriptions: create and delete subscriptions, get metadata about existing subscriptions, and list subscriptions.
• Topic stats: list producers and consumers for a topic through topic stats.
• Clusters: get cluster metadata, list clusters, as well as set and update cluster properties.

The pulsar-admin-go library provides several benefits for developers:

1. Unified Go API: developers can operate Pulsar resources using a unified Go API. This simplifies Pulsar management tasks by abstracting the underlying Pulsar admin HTTP operations.
2. Simplified Development: seamlessly integrates with other management tools like terraform-provider-pulsar, pulsar-resources-operator, and pulsarctl with the pulsar-admin-go library.
3. Improved Dependency Management: easier to control Go module dependencies and software releases.

II. Installing pulsar-admin-go Library

To use the pulsar-admin-go library, you need Go version 1.18 or higher and Go Modules enabled. To install the library, run the following command:

go get github.com/streamnative/pulsar-admin-go

III. Basic Usage

Here are some basic examples of how to use the pulsar-admin-go library.

To connect to a Pulsar cluster, you can create an Admin client with ServiceURL or Auth Token.

Create an Admin client with ServiceURL.

import (
    "github.com/streamnative/pulsar-admin-go"
)

func main() {
    cfg := &pulsaradmin.Config{
        WebServiceURL: "http://localhost:8080",
    }
    admin, err := pulsaradmin.NewClient(cfg)
    if err != nil {
        panic(err)
    }
}
}

Create an Admin client with Auth Token.

import (
    "github.com/streamnative/pulsar-admin-go"
)

func main() {
    cfg := &pulsaradmin.Config{
        // Use JWT Token for authentication and please note to allocate a token with Pulsar admin role
        // https://pulsar.apache.org/docs/2.11.x/security-jwt/#configure-jwt-authentication-in-pulsar-clients
        Token: "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJKb2UifQ.ipevRNuRP6HflG8cFKnmUPtypruRC4fb1DWtoLL62SY",
    }
    admin, err := pulsaradmin.NewClient(cfg)
    if err != nil {
        panic(err)
    }
}

Create a tenant, namespace, and topic.

import (
    "github.com/streamnative/pulsar-admin-go"
    "github.com/streamnative/pulsar-admin-go/pkg/utils"
)

func main() {
    cfg := &pulsaradmin.Config{}
    admin, err := pulsaradmin.NewClient(cfg)
    if err != nil {
        panic(err)
    }

    // Create a new tenant
    admin.Tenants().Create(utils.TenantData{
        Name: "new-tenant",
    })

    // Create a new namespace
    admin.Namespaces().CreateNamespace("new-tenant/new-namespace")

    // Create a new namespace with 3 partitions
    topic, _ := utils.GetTopicName("new-tenant/new-namespace/new-topic")
    admin.Topics().Create(*topic, 3)
}

IV. Advanced Usage

Here are some advanced examples of how to use the pulsar-admin-go library.

Configure geo-replication.

import (
    "github.com/streamnative/pulsar-admin-go"
    "github.com/streamnative/pulsar-admin-go/pkg/utils"
)

func main() {
    cfg := &pulsaradmin.Config{}
    admin, err := pulsaradmin.NewClient(cfg)
    if err != nil {
        panic(err)
    }

    // Deploy two pulsar clusters, and set the clusters array as your deployed pulsar clusters name
    // https://pulsar.apache.org/docs/2.11.x/install-deploy-upgrade-landing/
    // https://pulsar.apache.org/docs/2.11.x/administration-geo/
    clusters := []string{"us-west", "us-east"}

    admin.Tenants().Create(utils.TenantData{
        Name:            "geo-tenant",
        AllowedClusters: clusters,
    })

    admin.Namespaces().CreateNamespace("geo-tenant/geo-ns")
    admin.Namespaces().SetNamespaceReplicationClusters("geo-tenant/geo-ns", clusters)
}

Configure permissions for namespace and topic.

import (
   "github.com/streamnative/pulsar-admin-go"
   "github.com/streamnative/pulsar-admin-go/pkg/utils"
)

func main() {
   cfg := &pulsaradmin.Config{}

   admin, err := pulsaradmin.NewClient(cfg)
   if err != nil {
      panic(err)
   }

   // Configure admin permission for namespace public/default
   ns, _ := utils.GetNamespaceName("public/default")
   admin.Namespaces().GrantNamespacePermission(*ns, "admin", []utils.AuthAction{"produce", "consume"})

   // Configure admin permission for topic public/default/admin
   tp, _ := utils.GetTopicName("public/default/admin")
   admin.Topics().GrantPermission(*tp, "admin", []utils.AuthAction{"produce", "consume"})

‍

Configure retention policy for namespace.

import (
    "fmt"

    "github.com/streamnative/pulsar-admin-go"
    "github.com/streamnative/pulsar-admin-go/pkg/utils"
)

func main() {
    cfg := &pulsaradmin.Config{}
    admin, err := pulsaradmin.NewClient(cfg)
    if err != nil {
        panic(err)
    }

    // Create a new tenant
    admin.Tenants().Create(utils.TenantData{
        Name: "new-tenant",
    })

    // Create a new namespace
    admin.Namespaces().CreateNamespace("new-tenant/new-namespace")

    // Set the Retention Policy for this namespace
    admin.Namespaces().SetRetention("new-tenant/new-namespace", utils.RetentionPolicies{RetentionSizeInMB: 10240, RetentionTimeInMinutes: 180})

    // Get the Retention Policy for this namespace
    fmt.Println(admin.Namespaces().GetRetention("new-tenant/new-namespace"))
}

VI. Conclusion

The pulsar-admin-go library is a convenient way to manage Apache Pulsar clusters using Go. The library provides a set of intuitive interfaces that allow you to perform a wide range of tasks with ease. This library allows you to automate Pulsar management tasks, and integrate them into your applications. By using pulsar-admin-go, managing Pulsar clusters becomes easier and more efficient, allowing you to get the most out of this powerful messaging system. ‍

VII. More Resources

Join the Apache Pulsar community today and take part in shaping the future of messaging and streaming. Check out the GitHub repos, and documentation, and contribute to building an exciting project.