dcpelasticsearch

README

Go Dcp Elasticsearch

Go implementation of the Elasticsearch Connect Couchbase.

Go Dcp Elasticsearch streams documents from Couchbase Database Change Protocol (DCP) and writes to Elasticsearch index in near real-time. You can find more information by looking at our docs.

Features

• Less resource usage and higher throughput(see Benchmarks).
• Custom routing support(see Example).
• Update multiple documents for a DCP event(see Example).
• Handling different DCP events such as expiration, deletion and mutation(see Example).
• Elasticsearch compression request body support.
• Managing batch configurations such as maximum batch size, batch bytes, batch ticker durations.
• Scale up and down by custom membership algorithms(Couchbase, KubernetesHa, Kubernetes StatefulSet or Static, see examples).
• Easily manageable configurations.

Benchmarks

The benchmark was made with the 1,001,006 Couchbase document, because it is possible to more clearly observe the difference in the batch structure between the two packages. Default configurations for Java Elasticsearch Connect Couchbase used for both connectors.

Package	Time to Process Events	Elasticsearch Indexing Rate(/s)	Average CPU Usage(Core)	Average Memory Usage
Go Dcp Elasticsearch(Go 1.20)	50s		0.486	408MB
Java Elasticsearch Connect Couchbase(JDK15)	80s		0.31	1091MB

Example

Struct Config

func mapper(event couchbase.Event) []document.ESActionDocument {
  if event.IsMutated {
    e := document.NewIndexAction(event.Key, event.Value, nil)
    return []document.ESActionDocument{e}
  }
  e := document.NewDeleteAction(event.Key, nil)
  return []document.ESActionDocument{e}
}

func main() {
  connector, err := dcpelasticsearch.NewConnectorBuilder(config.Config{
    Elasticsearch: config.Elasticsearch{
      CollectionIndexMapping: map[string]string{
        "_default": "indexname",
      },
      Urls: []string{"http://localhost:9200"},
    },
    Dcp: dcpConfig.Dcp{
      Username:   "user",
      Password:   "password",
      BucketName: "dcp-test",
      Hosts:      []string{"localhost:8091"},
      Dcp: dcpConfig.ExternalDcp{
        Group: dcpConfig.DCPGroup{
          Name: "groupName",
        },
      },
      Metadata: dcpConfig.Metadata{
        Config: map[string]string{
          "bucket":     "checkpoint-bucket-name",
          "scope":      "_default",
          "collection": "_default",
        },
        Type: "couchbase",
      },
    },
  }).
    SetMapper(mapper).
    Build()
  if err != nil {
    panic(err)
  }

  defer connector.Close()
  connector.Start()
}

File Config

Default Mapper

Configuration

Dcp Configuration

Check out on go-dcp

Elasticsearch Specific Configuration

Variable	Type	Required	Default	Description
elasticsearch.collectionIndexMapping	map[string]string	yes		Defines which Couchbase collection events will be written to which index
elasticsearch.urls	[]string	yes		Elasticsearch connection urls
elasticsearch.username	string	no		The username of Elasticsearch
elasticsearch.password	string	no		The password of Elasticsearch
elasticsearch.typeName	string	no		Defines Elasticsearch index type name
elasticsearch.batchSizeLimit	int	no	1000	Maximum message count for batch, if exceed flush will be triggered.
elasticsearch.batchTickerDuration	time.Duration	no	10s	Batch is being flushed automatically at specific time intervals for long waiting messages in batch.
elasticsearch.batchCommitTickerDuration	time.Duration	no	0s	Configures checkpoint offset save time, By default, after batch flushing, the offsets are updated immediately, this period can be increased for performance.
elasticsearch.batchByteSizeLimit	int, string	no	10mb	Maximum size(byte) for batch, if exceed flush will be triggered. 10mb is default.
elasticsearch.maxConnsPerHost	int	no	512	Maximum number of connections per each host which may be established
elasticsearch.maxIdleConnDuration	time.Duration	no	10s	Idle keep-alive connections are closed after this duration.
elasticsearch.compressionEnabled	boolean	no	false	Compression can be used if message size is large, CPU usage may be affected.
elasticsearch.concurrentRequest	int	no	1	Concurrent bulk request count
elasticsearch.disableDiscoverNodesOnStart	boolean	no	false	Disable discover nodes when initializing the client.
elasticsearch.discoverNodesInterval	time.Duration	no	5m	Discover nodes periodically
elasticsearch.rejectionLog.index	string	no	cbes-rejects	Rejection log index name. cbes-rejects is default.
elasticsearch.rejectionLog.includeSource	boolean	no	false	Includes rejection log source info. false is default.
elasticsearch.maxRetries	int	no	math.MaxInt	Maximum retry count for the Elasticsearch client.

Exposed metrics

Metric Name	Description	Labels	Value Type
cbgo_elasticsearch_connector_latency_ms_current	Time to adding to the batch.	N/A	Gauge
cbgo_elasticsearch_connector_bulk_request_process_latency_ms_current	Time to process bulk request.	N/A	Gauge
cbgo_elasticsearch_connector_action_total_current	Count elasticsearch actions	action_type: Type of action (e.g., delete, index) result: Result of the action (e.g., success, error) index_name: The name of the index to which the action is applied	Counter

You can also use all DCP-related metrics explained here. All DCP-related metrics are automatically injected. It means you don't need to do anything.

Grafana Metric Dashboard

Grafana & Prometheus Example

Contributing

Go Dcp Elasticsearch is always open for direct contributions. For more information please check our Contribution Guideline document.

License

Released under the MIT License.

Documentation

Index

• func DefaultMapper(event couchbase.Event) []document.ESActionDocument
• type Connector
• type ConnectorBuilder
  • func NewConnectorBuilder(config any) *ConnectorBuilder
  • func (c *ConnectorBuilder) Build() (Connector, error)
  • func (c *ConnectorBuilder) SetLogger(logrus *logrus.Logger) *ConnectorBuilder
  • func (c *ConnectorBuilder) SetMapper(mapper Mapper) *ConnectorBuilder
  • func (c *ConnectorBuilder) SetMetricCollectors(collectors ...prometheus.Collector)
  • func (c *ConnectorBuilder) SetSinkResponseHandler(sinkResponseHandler dcpElasticsearch.SinkResponseHandler) *ConnectorBuilder
• type DcpEventHandler
  • func (h *DcpEventHandler) AfterRebalanceEnd()
  • func (h *DcpEventHandler) AfterRebalanceStart()
  • func (h *DcpEventHandler) AfterStreamStart()
  • func (h *DcpEventHandler) AfterStreamStop()
  • func (h *DcpEventHandler) BeforeRebalanceEnd()
  • func (h *DcpEventHandler) BeforeRebalanceStart()
  • func (h *DcpEventHandler) BeforeStreamStart()
  • func (h *DcpEventHandler) BeforeStreamStop()
• type Mapper

Functions

func DefaultMapper

func DefaultMapper(event couchbase.Event) []document.ESActionDocument

Types

type Connector

type Connector interface {
	Start()
	Close()
	GetDcpClient() dcpCouchbase.Client
}

type ConnectorBuilder

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

func NewConnectorBuilder

func NewConnectorBuilder(config any) *ConnectorBuilder

func (*ConnectorBuilder) Build

func (c *ConnectorBuilder) Build() (Connector, error)

func (*ConnectorBuilder) SetLogger

func (c *ConnectorBuilder) SetLogger(logrus *logrus.Logger) *ConnectorBuilder

func (*ConnectorBuilder) SetMapper

func (c *ConnectorBuilder) SetMapper(mapper Mapper) *ConnectorBuilder

func (*ConnectorBuilder) SetMetricCollectors

func (c *ConnectorBuilder) SetMetricCollectors(collectors ...prometheus.Collector)

func (*ConnectorBuilder) SetSinkResponseHandler

func (c *ConnectorBuilder) SetSinkResponseHandler(sinkResponseHandler dcpElasticsearch.SinkResponseHandler) *ConnectorBuilder

type DcpEventHandler

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

func (*DcpEventHandler) AfterRebalanceEnd

func (h *DcpEventHandler) AfterRebalanceEnd()

func (*DcpEventHandler) AfterRebalanceStart

func (h *DcpEventHandler) AfterRebalanceStart()

func (*DcpEventHandler) AfterStreamStart

func (h *DcpEventHandler) AfterStreamStart()

func (*DcpEventHandler) AfterStreamStop

func (h *DcpEventHandler) AfterStreamStop()

func (*DcpEventHandler) BeforeRebalanceEnd

func (h *DcpEventHandler) BeforeRebalanceEnd()

func (*DcpEventHandler) BeforeRebalanceStart

func (h *DcpEventHandler) BeforeRebalanceStart()

func (*DcpEventHandler) BeforeStreamStart

func (h *DcpEventHandler) BeforeStreamStart()

func (*DcpEventHandler) BeforeStreamStop

func (h *DcpEventHandler) BeforeStreamStop()

type Mapper

type Mapper func(event couchbase.Event) []document.ESActionDocument