output

README

benthos-cassandra-sink

A simple Cassandra sink for Benthos.

Installation

$ go get -u github.com/codingconcepts/benthos-cassandra-sink

Usage

Configure a simple Go application that references this repository and executes its init function:

package main

import (
	"github.com/Jeffail/benthos/v3/lib/service"

	_ "github.com/codingconcepts/benthos-cassandra-sink"
)

func main() {
	service.Run()
}

Configure a Benthos configuration file:

input:
    stdout: {}

output:
    type: cassandra
    plugin:
        nodes:
        - "127.0.0.1"
        keyspace: YOUR_KEYSPACE
        table: YOUR_TABLE
        password_authenticator:
            enabled: true
            username: USERNAME
            password: PASSWORD

Thanks

• Adam Wasila for the inspiration from https://github.com/adamwasila/benthos/tree/WIP_output_cassandra
• Ash Jeffs for the help in getting this output to work!

Documentation

Index

• type Cassandra
  • func NewCassandra(conf CassandraConfig, log log.Modular, stats metrics.Type) (*Cassandra, error)
  • func (c *Cassandra) CloseAsync()
  • func (c *Cassandra) ConnectWithContext(ctx context.Context) error
  • func (c *Cassandra) Connected() bool
  • func (c *Cassandra) WaitForClose(timeout time.Duration) error
  • func (c *Cassandra) WriteWithContext(ctx context.Context, msg types.Message) error
• type CassandraConfig
  • func NewCassandraConfig() *CassandraConfig
• type PasswordAuthenticator

Types

type Cassandra

type Cassandra struct {
	// contains filtered or unexported fields
}

Cassandra holds the configuration properties for a running instance of the Cassandra sink.

func NewCassandra

func NewCassandra(conf CassandraConfig, log log.Modular, stats metrics.Type) (*Cassandra, error)

NewCassandra creates a new Cassandra writer type.

func (*Cassandra) CloseAsync

func (c *Cassandra) CloseAsync()

CloseAsync shuts down the Cassandra output and stops processing messages.

func (*Cassandra) ConnectWithContext

func (c *Cassandra) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to an Cassandra.

func (*Cassandra) Connected

func (c *Cassandra) Connected() bool

Connected returns true if the connection to Cassanrda is open.

func (*Cassandra) WaitForClose

func (c *Cassandra) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Cassandra output has closed down.

func (*Cassandra) WriteWithContext

func (c *Cassandra) WriteWithContext(ctx context.Context, msg types.Message) error

WriteWithContext writes a message to Cassandra.

type CassandraConfig

type CassandraConfig struct {
	Nodes                 []string              `json:"nodes" yaml:"nodes"`
	AsyncParts            bool                  `json:"async" yaml:"async"`
	PasswordAuthenticator PasswordAuthenticator `json:"password_authenticator" yaml:"password_authenticator"`
	Keyspace              string                `json:"keyspace" yaml:"keyspace"`
	Table                 string                `json:"table" yaml:"table"`
	Consistency           string                `json:"consistency" yaml:"consistency"`
	retries.Config        `json:",inline" yaml:",inline"`
}

CassandraConfig contains configuration fields for the Cassandra output type.

func NewCassandraConfig

func NewCassandraConfig() *CassandraConfig

NewCassandraConfig creates a new CassandraConfig with default values.

type PasswordAuthenticator

type PasswordAuthenticator struct {
	Enabled  bool   `json:"enabled" yaml:"enabled"`
	Username string `json:"username" yaml:"username"`
	Password string `json:"password" yaml:"password"`
}

PasswordAuthenticator contains the fields that will be used to authenticate with the Cassandra cluster.