gremcos

README

gremcos

Gremcos is a fork of schwartzmx/gremtune with alterations to make it compatible with Gremlin API of Azure Cosmos DB which is a Graph Database (Gremlin API) for Azure.

Gremcos is a fast, efficient, and easy-to-use client for the TinkerPop graph database stack. It is a gremlin language driver which uses WebSockets to interface with gremlin server and has a strong emphasis on concurrency and scalability. Please keep in mind that gremcos is still under heavy development and although effort is being made to fully cover gremcos with reliable tests, bugs may be present in several areas.

Installation

go get github.com/supplyon/gremcos

Examples

• See: examples/README.md

Hints

Response Format

This implementation supports Graphson 2.0 (not 3) in order to be compatible to CosmosDB. This means all the responses from the CosmosDB server as well as the responses from the local gremlin-server have to comply with the 2.0 format.

Azure Cosmos Gremlin Implementation Differences

Modifications where made to gremtune in order to be compliant to Azure Cosmos DB. Differences in gremlin support can be found at: Azure Cosmos DB Gremlin compatibility

This implementation is only working/ compatible with TinkerPop 3.4.0.

Cosmos DB specific error handling is done and described at ErrorHandling.md. For example error responses returned by Cosmos due to a usage rate limit violation are handled accordingly.

Local Development

For being able to develop locally against a local graph data base one can start a local gremlin-server via make infra.up. In order to be able to use all features the query language has to be switched to QueryLanguageTinkerpopGremlin.

Switch the Query Language

Since the query language of the Cosmos DB and the tinkerpop gremlin implementation are not 100% compatible it is possible to set the language based on the use-case. The following piece of code depicts how to set the query language.

    // If you want to run your queries against a apache tinkerpop gremlin server it is recommended
    // to switch the used query language to QueryLanguageTinkerpopGremlin.
    // Per default the CosmosDB compatible query language will be used.
    api.SetQueryLanguageTo(api.QueryLanguageTinkerpopGremlin)

License

See LICENSE

3rd Party Licenses

• difflib license

Documentation

Index

• Variables
• func Dial(conn interfaces.Dialer, errorChannel chan error, options ...clientOption) (*client, error)
• func IsNetworkErr(err error) bool
• func NewPool(createQueryExecutor QueryExecutorFactoryFunc, maxActiveConnections int, ...) (*pool, error)
• func NewWebsocket(host string, options ...optionWebsocket) (interfaces.Dialer, error)
• func PingInterval(interval time.Duration) clientOption
• func SetAuth(credentialProvider CredentialProvider) clientOption
• func SetBufferSize(readBufferSize int, writeBufferSize int) optionWebsocket
• func SetReadingWait(wait time.Duration) optionWebsocket
• func SetTimeout(timeout time.Duration) optionWebsocket
• func SetWritingWait(wait time.Duration) optionWebsocket
• func WithMetrics(metrics clientMetrics) clientOption
• type Cosmos
  • func New(host string, options ...Option) (Cosmos, error)
• type CredentialProvider
• type Error
  • func (e Error) Error() string
• type ErrorCategory
• type Metrics
  • func NewMetrics(namespace string) *Metrics
• type Option
  • func AutomaticRetries(maxRetries int, timeout time.Duration) Option
  • func ConnectionIdleTimeout(timeout time.Duration) Option
  • func MetricsPrefix(prefix string) Option
  • func NumMaxActiveConnections(numMaxActiveConnections int) Option
  • func QueryTimeouts(readTimeout time.Duration, writeTimeout time.Duration) Option
  • func WithAuth(username string, password string) Option
  • func WithLogger(logger zerolog.Logger) Option
  • func WithResourceTokenAuth(credentialProvider CredentialProvider) Option
• type QueryExecutorFactoryFunc
• type StaticCredentialProvider
  • func (c StaticCredentialProvider) Password() (string, error)
  • func (c StaticCredentialProvider) Username() (string, error)

Variables

var ErrNoConnection = Error{Wrapped: fmt.Errorf("no connection"), Category: ErrorCategoryConnectivity}

var MimeType = []byte("application/vnd.gremlin-v2.0+json")

MimeType used for communication with the gremlin server.

Functions

func Dial

func Dial(conn interfaces.Dialer, errorChannel chan error, options ...clientOption) (*client, error)

Dial returns a client for interaction with the Gremlin Server specified in the host IP. The client is already connected.

func IsNetworkErr

func IsNetworkErr(err error) bool

IsNetworkErr determines whether the given error is related to any network issues (timeout, connectivity,..)

func NewPool

func NewPool(createQueryExecutor QueryExecutorFactoryFunc, maxActiveConnections int, idleTimeout time.Duration, logger zerolog.Logger) (*pool, error)

NewPool creates a new pool which is a QueryExecutor

func NewWebsocket

func NewWebsocket(host string, options ...optionWebsocket) (interfaces.Dialer, error)

NewWebsocket returns a WebSocket dialer to use when connecting to Gremlin Server

func PingInterval

func PingInterval(interval time.Duration) clientOption

PingInterval sets the ping interval, which is the interval to send the ping frame to the peer

func SetAuth

func SetAuth(credentialProvider CredentialProvider) clientOption

SetAuth sets credentials provider for an authenticated connection

func SetBufferSize

func SetBufferSize(readBufferSize int, writeBufferSize int) optionWebsocket

SetBufferSize sets the read/write buffer size

func SetReadingWait

func SetReadingWait(wait time.Duration) optionWebsocket

SetReadingWait sets the time for waiting that reading occur

func SetTimeout

func SetTimeout(timeout time.Duration) optionWebsocket

SetTimeout sets the dial handshake timeout

func SetWritingWait

func SetWritingWait(wait time.Duration) optionWebsocket

SetWritingWait sets the time for waiting that writing occur

func WithMetrics

func WithMetrics(metrics clientMetrics) clientOption

WithMetrics sets the metrics provider

Types

type Cosmos

type Cosmos interface {
	// ExecuteQuery executes the given query and returns the according responses from the CosmosDB
	ExecuteQuery(query interfaces.QueryBuilder) ([]interfaces.Response, error)

	// Execute can be used to execute a raw query (string). This can be used to issue queries that are not yet supported by the QueryBuilder.
	Execute(query string) ([]interfaces.Response, error)

	// ExecuteAsync can be used to issue a query and streaming in the responses as they are available / are provided by the CosmosDB
	ExecuteAsync(query string, responseChannel chan interfaces.AsyncResponse) (err error)

	// ExecuteWithBindings can be used to execute a raw query (string) with optional bindings/rebindings. This can be used to issue queries that are not yet supported by the QueryBuilder.
	ExecuteWithBindings(path string, bindings, rebindings map[string]interface{}) (resp []interfaces.Response, err error)

	// IsConnected returns true in case the connection to the CosmosDB is up, false otherwise.
	IsConnected() bool

	// Stop stops the connector, terminates all background go routines and closes open connections.
	Stop() error

	// String returns a string representation of the cosmos connector
	String() string

	// IsHealthy returns nil in case the connection to the CosmosDB is up, the according error otherwise.
	IsHealthy() error
}

Cosmos is an abstraction of the CosmosDB

func New

func New(host string, options ...Option) (Cosmos, error)

New creates a new instance of the Cosmos (-DB connector)

type CredentialProvider

type CredentialProvider interface {
	Username() (string, error)
	Password() (string, error)
}

CredentialProvider provides access to cosmos credentials. In order to be able to provide dynamic credentials aka cosmos resource tokens you have to implement this interface and ensure in this implementation that always a valid resource token is returned by Password().

type Error

type Error struct {
	Wrapped  error
	Category ErrorCategory
}

func (Error) Error

func (e Error) Error() string

type ErrorCategory

type ErrorCategory string

const (
	ErrorCategoryGeneral      ErrorCategory = "GeneralErr"
	ErrorCategoryConnectivity ErrorCategory = "ConnectivityErr"
	ErrorCategoryAuth         ErrorCategory = "AuthErr"
	ErrorCategoryClient       ErrorCategory = "ClientErr"
	ErrorCategoryServer       ErrorCategory = "ServerErr"
)

type Metrics

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

Metrics represents the collection of metrics internally set by the service.

func NewMetrics

func NewMetrics(namespace string) *Metrics

NewMetrics returns the metrics collection

type Option

type Option func(*cosmosImpl)

Option is the struct for defining optional parameters for Cosmos

func AutomaticRetries

func AutomaticRetries(maxRetries int, timeout time.Duration) Option

AutomaticRetries tries to retry failed requests, if appropriate. Retries are limited to maxRetries. Retrying is stopped after timeout is reached. Appropriate error codes are 409, 412, 429, 1007, 1008 see https://docs.microsoft.com/en-us/azure/cosmos-db/graph/gremlin-headers#status-codes Hint: Be careful when specifying the values for maxRetries and timeout. They influence how much latency is added on requests that need to be retried.

For example if maxRetries = 1 and timeout = 1s the call might take 1s longer to return a potential persistent error.

func ConnectionIdleTimeout

func ConnectionIdleTimeout(timeout time.Duration) Option

ConnectionIdleTimeout specifies the timeout after which idle connections will be removed from the internal connection pool

func MetricsPrefix

func MetricsPrefix(prefix string) Option

MetricsPrefix can be used to customize the metrics prefix as needed for a specific service. Per default 'gremcos' is used as prefix.

func NumMaxActiveConnections

func NumMaxActiveConnections(numMaxActiveConnections int) Option

NumMaxActiveConnections specifies the maximum amount of active connections.

func QueryTimeouts

func QueryTimeouts(readTimeout time.Duration, writeTimeout time.Duration) Option

QueryTimeouts specifies the timeouts for executing a query. readTimeout specifies the amount of time a query can last until the response is completely fetched at the client. writeTimeout specifies the amount of time its allowed to take to send the query and all related data to the server.

func WithAuth

func WithAuth(username string, password string) Option

WithAuth sets credentials for an authenticated connection using static credentials (primary-/ secondary cosmos key as password)

func WithLogger

func WithLogger(logger zerolog.Logger) Option

WithLogger specifies the logger to use

func WithResourceTokenAuth

func WithResourceTokenAuth(credentialProvider CredentialProvider) Option

WithResourceTokenAuth sets credential provider that is used to authenticate the requests to cosmos. With this approach dynamic credentials (cosmos resource tokens) can be used for authentication. To do this you have to provide a CredentialProvider implementation that takes care for providing a valid (not yet expired) resource token

myResourceTokenProvider := MyDynamicCredentialProvider{}
New("wss://example.com", WithResourceTokenAuth(myResourceTokenProvider))

If you want to use static credentials (primary-/ secondary cosmos key as password) instead you can either use "WithAuth".

New("wss://example.com", WithAuth("username","primary-key"))

Or you use the default implementation for a static credential provider "StaticCredentialProvider"

staticCredProvider := StaticCredentialProvider{UsernameStatic: "username", PasswordStatic: "primary-key"}
New("wss://example.com", WithResourceTokenAuth(staticCredProvider))

type QueryExecutorFactoryFunc

type QueryExecutorFactoryFunc func() (interfaces.QueryExecutor, error)

type StaticCredentialProvider

type StaticCredentialProvider struct {
	UsernameStatic string
	PasswordStatic string
}

StaticCredentialProvider is a default implementation of the CredentialProvider interface. It can be used in case you have no dynamic credentials but use the static primary-/ secondary cosmos key.

func (StaticCredentialProvider) Password

func (c StaticCredentialProvider) Password() (string, error)

func (StaticCredentialProvider) Username

func (c StaticCredentialProvider) Username() (string, error)