sdk

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 5, 2022 License: Apache-2.0 Imports: 17 Imported by: 371

README

Conduit Connector SDK

License Build Go Report Card Go Reference

This repository contains the Go software development kit for implementing a connector for Conduit. If you want to implement a connector in another language please have a look at the connector protocol.

Quickstart

Create a new folder and initialize a fresh go module:

go mod init example.com/conduit-connector-demo

Add the connector SDK dependency:

go get github.com/conduitio/conduit-connector-sdk

With this you can start implementing the connector. To implement a source (a connector that reads from a 3rd party resource and sends data to Conduit) create a struct that implements sdk.Source. To implement a destination (a connector that receives data from Conduit and writes it to a 3rd party resource) create a struct that implements sdk.Destination. You can implement both to make a connector that can be used both as a source or a destination.

Apart from the source and/or destination you should create constructor functions that return a sdk.Source, sdk.Destination and sdk.Specification respectively.

The last part is the entrypoint, it needs to call sdk.Serve and pass in the constructor functions mentioned before. If the connector does not implement a source or destination you should pass in nil instead.

package main

import (
	demo "example.com/conduit-connector-demo"
	sdk "github.com/conduitio/conduit-connector-sdk"
)

func main() {
	sdk.Serve(
		demo.Specification,  // func Specification() sdk.Specification { ... }
		demo.NewSource,      // func NewSource() sdk.Source { ... }
		demo.NewDestination, // func NewDestination() sdk.Destination { ... }
	)
}

Now you can build the standalone connector:

go build path/to/main.go

You will get a compiled binary which Conduit can use as a connector. To run your connector as part of a Conduit pipeline you can create it using the connectors API and specify the path to the compiled connector binary in the field plugin.

Here is an example request to POST /v1/connectors (find more about the Conduit API):

{
  "type": "TYPE_SOURCE",
  "plugin": "/path/to/compiled/connector/binary",
  "pipelineId": "...",
  "config": {
    "name": "my-connector",
    "settings": {
      "my-key": "my-value"
    }
  }
}

FAQ

Q: Is there a standard format for errors?

Conduit doesn't expect any specific error format. We still encourage developers to follow the conventional error message formatting and include enough contextual information to make debugging as easy as possible (e.g. stack trace, information about the value that caused the error, internal state).

Q: Is there a standard format for logging?

Developers should use sdk.Logger to retrieve a *zerolog.Logger instance. It can be used to emit structured and leveled log messages that will be included in Conduit logs.

Keep in mind that logging in the hot path (e.g. reading or writing a record) can have a negative impact on performance and should be avoided. If you really want to add a log message in the hot path please use the "trace" level.

Q: Do I need to worry about ordering?

In case of the destination connector you do not have to worry about ordering. Conduit will supply records one by one in the order they were produced in the source.

On the other hand, the source connector is in charge of producing records and thus dictates the order. That said, you do not have to worry about concurrent reads, the SDK will call Source.Read repeatedly and only in one goroutine, all you have to do is return one record at a time.

Examples

For examples of simple connectors you can look at existing connectors like conduit-connector-generator or conduit-connector-file.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrBackoffRetry can be returned by Source.Read to signal the SDK there is
	// no record to fetch right now and it should try again later.
	ErrBackoffRetry = errors.New("backoff retry")

	// ErrUnimplemented is returned in functions of plugins that don't implement
	// a certain method.
	ErrUnimplemented = errors.New("action not implemented")
)

Functions

func Logger 

func Logger(ctx context.Context) *zerolog.Logger

Logger returns an instance of a logger that can be used for leveled and structured logging in a plugin. The logger will respect the log level configured in Conduit.

func NewDestinationPlugin 

func NewDestinationPlugin(impl Destination) cpluginv1.DestinationPlugin

NewDestinationPlugin takes a Destination and wraps it into an adapter that converts it into a cpluginv1.DestinationPlugin. If the parameter is nil it will wrap UnimplementedDestination instead.

func NewSourcePlugin 

func NewSourcePlugin(impl Source) cpluginv1.SourcePlugin

NewSourcePlugin takes a Source and wraps it into an adapter that converts it into a cpluginv1.SourcePlugin. If the parameter is nil it will wrap UnimplementedSource instead.

func NewSpecifierPlugin 

func NewSpecifierPlugin(specs Specification) cpluginv1.SpecifierPlugin

NewSpecifierPlugin takes a Specification and wraps it into an adapter that converts it into a cpluginv1.SpecifierPlugin.

func Serve 

func Serve(
	specFactory func() Specification,
	sourceFactory func() Source,
	destinationFactory func() Destination,
)

Serve starts the plugin and takes care of its whole lifecycle by blocking until the plugin can safely stop running. Any fixable errors will be output to os.Stderr and the process will exit with a status code of 1. Serve will panic for unexpected conditions where a user's fix is unknown.

It is essential that nothing gets written to stdout or stderr before this function is called, as the first output is used to perform the initial handshake.

Plugins should call Serve in their main() functions.

Types

type AckFunc 

type AckFunc func(error) error

AckFunc is a function that forwards the acknowledgment to Conduit. It returns an error in case the forwarding failed, otherwise it returns nil. If nil is passed to this function the record is assumed to be successfully processed, otherwise the parameter should contain an error describing why a record failed to be processed. If an error is returned from AckFunc it should be regarded as fatal and all processing should be stopped as soon as possible.

type Data 

type Data interface {
	Bytes() []byte
	// contains filtered or unexported methods
}

Data is a structure that contains some bytes. The only structs implementing Data are RawData and StructuredData.

type Destination 

type Destination interface {
	// Configure is the first function to be called in a connector. It provides the
	// connector with the configuration that needs to be validated and stored.
	// In case the configuration is not valid it should return an error.
	// Testing if your connector can reach the configured data source should be
	// done in Open, not in Configure.
	Configure(context.Context, map[string]string) error

	// Open is called after Configure to signal the plugin it can prepare to
	// start writing records. If needed, the plugin should open connections in
	// this function.
	Open(context.Context) error

	// WriteAsync receives a Record and can cache it internally and write it at
	// a later time. Once the record is successfully written to the destination
	// the plugin must call the provided AckFunc function with a nil error. If
	// the plugin failed to write the record to the destination it must call the
	// supplied AckFunc with a non-nil error. If any AckFunc is left uncalled
	// the connector will not be able to exit gracefully. The WriteAsync
	// function should only return an error in case a critical error happened,
	// it is expected that all future calls to WriteAsync will fail with the
	// same error.
	// If the plugin caches records before writing them to the destination it
	// needs to store the AckFunc as well and call it once the record is
	// written.
	// AckFunc will panic if it's called more than once.
	// If WriteAsync returns ErrUnimplemented the SDK will fall back and call
	// Write instead.
	WriteAsync(context.Context, Record, AckFunc) error

	// Write receives a Record and is supposed to write the record to the
	// destination right away. If the function returns nil the record is assumed
	// to have reached the destination.
	// WriteAsync takes precedence and will be tried first to write a record. If
	// WriteAsync is not implemented the SDK will fall back and use Write
	// instead.
	Write(context.Context, Record) error

	// Flush signals the plugin it should flush any cached records and call all
	// outstanding AckFunc functions. This function needs to be implemented only
	// if the struct implements WriteAsync. No more calls to WriteAsync
	// will be issued after Flush is called.
	Flush(context.Context) error

	// Teardown signals to the plugin that all records were written and there
	// will be no more calls to any other function. After Teardown returns, the
	// plugin should be ready for a graceful shutdown.
	Teardown(context.Context) error
	// contains filtered or unexported methods
}

Destination receives records from Conduit and writes them to 3rd party resources. All implementations must embed UnimplementedDestination for forward compatibility. When implementing Destination you can choose between implementing the method WriteAsync or Write. If both are implemented, then WriteAsync will be used. Use WriteAsync if the plugin will cache records and write them to the 3rd party resource in batches, otherwise use Write.

type MockSource 

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

MockSource is a mock of Source interface.

func NewMockSource 

func NewMockSource(ctrl *gomock.Controller) *MockSource

NewMockSource creates a new mock instance.

func (*MockSource) Ack 

func (m *MockSource) Ack(arg0 context.Context, arg1 Position) error

Ack mocks base method.

func (*MockSource) Configure 

func (m *MockSource) Configure(arg0 context.Context, arg1 map[string]string) error

Configure mocks base method.

func (*MockSource) EXPECT 

func (m *MockSource) EXPECT() *MockSourceMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockSource) Open 

func (m *MockSource) Open(arg0 context.Context, arg1 Position) error

Open mocks base method.

func (*MockSource) Read 

func (m *MockSource) Read(arg0 context.Context) (Record, error)

Read mocks base method.

func (*MockSource) Teardown 

func (m *MockSource) Teardown(arg0 context.Context) error

Teardown mocks base method.

type MockSourceMockRecorder 

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

MockSourceMockRecorder is the mock recorder for MockSource.

func (*MockSourceMockRecorder) Ack 

func (mr *MockSourceMockRecorder) Ack(arg0, arg1 interface{}) *gomock.Call

Ack indicates an expected call of Ack.

func (*MockSourceMockRecorder) Configure 

func (mr *MockSourceMockRecorder) Configure(arg0, arg1 interface{}) *gomock.Call

Configure indicates an expected call of Configure.

func (*MockSourceMockRecorder) Open 

func (mr *MockSourceMockRecorder) Open(arg0, arg1 interface{}) *gomock.Call

Open indicates an expected call of Open.

func (*MockSourceMockRecorder) Read 

func (mr *MockSourceMockRecorder) Read(arg0 interface{}) *gomock.Call

Read indicates an expected call of Read.

func (*MockSourceMockRecorder) Teardown 

func (mr *MockSourceMockRecorder) Teardown(arg0 interface{}) *gomock.Call

Teardown indicates an expected call of Teardown.

type Parameter 

type Parameter struct {
	// Default is the default value of the parameter, if any.
	Default string
	// Required is whether it must be provided in the Config or not.
	Required bool
	// Description holds a description of the field and how to configure it.
	Description string
}

Parameter is a helper struct for defining plugin Specifications.

type Position 

type Position []byte

Position is a unique identifier for a record. It is the responsibility of the Source to choose and assign record positions, it can freely choose a format that makes sense and contains everything needed to restart a pipeline at a certain position.

type RawData 

type RawData []byte

RawData contains unstructured data in form of a byte slice.

func (RawData) Bytes 

func (d RawData) Bytes() []byte

Bytes simply casts RawData to a byte slice.

type Record 

type Record struct {
	// Position uniquely represents the record.
	Position Position
	// Metadata contains additional information regarding the record.
	Metadata map[string]string
	// CreatedAt represents the time when the change occurred in the source
	// system. If that's impossible to find out, then it should be the time the
	// change was detected by the connector.
	CreatedAt time.Time
	// Key represents a value that should be the same for records originating
	// from the same source entry (e.g. same DB row). In a destination Key will
	// never be null.
	Key Data
	// Payload holds the actual information that the record is transmitting. In
	// a destination Payload will never be null.
	Payload Data
}

Record represents a single data record produced by a source and/or consumed by a destination connector.

type Source 

type Source interface {
	// Configure is the first function to be called in a connector. It provides the
	// connector with the configuration that needs to be validated and stored.
	// In case the configuration is not valid it should return an error.
	// Testing if your connector can reach the configured data source should be
	// done in Open, not in Configure.
	Configure(context.Context, map[string]string) error

	// Open is called after Configure to signal the plugin it can prepare to
	// start producing records. If needed, the plugin should open connections in
	// this function. The position parameter will contain the position of the
	// last record that was successfully processed, Source should therefore
	// start producing records after this position. The context passed to Open
	// will be cancelled once the plugin receives a stop signal from Conduit.
	Open(context.Context, Position) error

	// Read returns a new Record and is supposed to block until there is either
	// a new record or the context gets cancelled. It can also return the error
	// ErrBackoffRetry to signal to the SDK it should call Read again with a
	// backoff retry.
	// If Read receives a cancelled context or the context is cancelled while
	// Read is running it must stop retrieving new records from the source
	// system and start returning records that have already been buffered. If
	// there are no buffered records left Read must return the context error to
	// signal a graceful stop. If Read returns ErrBackoffRetry while the context
	// is cancelled it will also signal that there are no records left and Read
	// won't be called again.
	// After Read returns an error the function won't be called again (except if
	// the error is ErrBackoffRetry, as mentioned above).
	// Read can be called concurrently with Ack.
	Read(context.Context) (Record, error)
	// Ack signals to the implementation that the record with the supplied
	// position was successfully processed. This method might be called after
	// the context of Read is already cancelled, since there might be
	// outstanding acks that need to be delivered. When Teardown is called it is
	// guaranteed there won't be any more calls to Ack.
	// Ack can be called concurrently with Read.
	Ack(context.Context, Position) error

	// Teardown signals to the plugin that there will be no more calls to any
	// other function. After Teardown returns, the plugin should be ready for a
	// graceful shutdown.
	Teardown(context.Context) error
	// contains filtered or unexported methods
}

Source fetches records from 3rd party resources and sends them to Conduit. All implementations must embed UnimplementedSource for forward compatibility.

type Specification 

type Specification struct {
	// Name is the name of the plugin.
	Name string
	// Summary is a brief description of the plugin and what it does.
	Summary string
	// Description is a more long form area appropriate for README-like text
	// that the author can provide for documentation about the specified
	// Parameters.
	Description string
	// Version string. Should be prepended with `v` like Go, e.g. `v1.54.3`
	Version string
	// Author declares the entity that created or maintains this plugin.
	Author string
	// DestinationParams and SourceParams is a map of named Parameters that
	// describe how to configure the plugin's Destination or Source.
	DestinationParams map[string]Parameter
	SourceParams      map[string]Parameter
}

Specification is returned by a plugin when Specify is called. It contains information about the configuration parameters for plugins and allows them to describe their parameters.

type StructuredData 

type StructuredData map[string]interface{}

StructuredData contains data in form of a map with string keys and arbitrary values.

func (StructuredData) Bytes 

func (d StructuredData) Bytes() []byte

Bytes returns the JSON encoding of the map.

type UnimplementedDestination 

type UnimplementedDestination struct{}

UnimplementedDestination should be embedded to have forward compatible implementations.

func (UnimplementedDestination) Configure 

func (UnimplementedDestination) Configure(context.Context, map[string]string) error

Configure needs to be overridden in the actual implementation.

func (UnimplementedDestination) Flush 

func (UnimplementedDestination) Flush(context.Context) error

Flush needs to be overridden in the actual implementation if WriteAsync is implemented, otherwise it is optional.

func (UnimplementedDestination) Open 

func (UnimplementedDestination) Open(context.Context) error

Open needs to be overridden in the actual implementation.

func (UnimplementedDestination) Teardown 

func (UnimplementedDestination) Teardown(context.Context) error

Teardown needs to be overridden in the actual implementation.

func (UnimplementedDestination) Write 

func (UnimplementedDestination) Write(context.Context, Record) error

Write needs to be overridden in the actual implementation, unless WriteAsync is implemented.

func (UnimplementedDestination) WriteAsync 

func (UnimplementedDestination) WriteAsync(context.Context, Record, AckFunc) error

WriteAsync needs to be overridden in the actual implementation, unless Write is implemented.

type UnimplementedSource 

type UnimplementedSource struct{}

UnimplementedSource should be embedded to have forward compatible implementations.

func (UnimplementedSource) Ack 

func (UnimplementedSource) Ack(context.Context, Position) error

Ack should be overridden if acks need to be forwarded to the source, otherwise it is optional.

func (UnimplementedSource) Configure 

func (UnimplementedSource) Configure(context.Context, map[string]string) error

Configure needs to be overridden in the actual implementation.

func (UnimplementedSource) Open 

func (UnimplementedSource) Open(context.Context, Position) error

Open needs to be overridden in the actual implementation.

func (UnimplementedSource) Read 

func (UnimplementedSource) Read(context.Context) (Record, error)

Read needs to be overridden in the actual implementation.

func (UnimplementedSource) Teardown 

func (UnimplementedSource) Teardown(context.Context) error

Teardown needs to be overridden in the actual implementation.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.