deckard

package module
v0.0.16 Latest Latest
Warning

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

 Go to latest Published: Jun 27, 2023 License: MIT Imports: 9 Imported by: 0

README

Deckard: A Highly Scalable Cyclic Priority Queue

License codecov Go Report Card

Artifact Hub Docker

Nuget org Golang

Slack

Deckard is a priority queue system inspired by projects such as Google Cloud PubSub, Nats, Kafka, and others. Its main distinction lies in its ability to associate a priority with each message and have a queue that can be optionally cyclic. This means that messages can be delivered again after a user-managed time. Additionally, Deckard implements a locking mechanism to prevent message processing for a specified duration.

deckard

Briefly:

  • An application inserts a message to be queued and its configuration (TTL, metadata, payload, etc).
    • The message will be prioritized with a default timestamp-based algorithm. The priority can also be provided by the application.
  • A worker application pull messages from Deckard at regular intervals and performs any processing.
    • When it finishes processing a message, the application must notify with the processing result.
    • When notifying, the application may provide a lock time, to lock the message for a certain duration of time before being requeued and delivered again.
    • It is also possible to notify a message changing its priority.
  • When the message's TTL is reached, it stops being delivered;
    • For some use cases the TTL can be set as infinite.
    • An application can also remove the message when notifying.

Motivation

Deckard was born from an initiative to simplify and unify applications called Frontiers which were part of STILINGUE's orchestration system for data gathering.

Several different implementations were made, but they proved to have limitations such as:

  • Debugging
  • Genericity
  • Scalability
  • Auditability
  • Observability
  • Prioritization
  • Developer friendliness

Deckard was created to solve these problems and to be used in any application that needs to queue data and prioritize it.

The main objectives of the project are:

  • Generic: any data and application should be able to use Deckard to queue messages;
  • Observable: it should be easy to visualize what happened in a request and easy to investigate problems with audit;
  • Developer friendly: easy to understand and pleasant to use;
  • Scalable: it should be possible to support thousands of requests per second and millions of messages;
What Deckard is not?

It is important to note that Deckard is not a conventional messaging/queue system. If your use case does not involve priority, cyclic queuing, or a locking mechanism, it is recommended to consider alternatives such as GCP PubSub, Kafka, RabbitMQ, Azure Service Bus, Amazon SQS, or other messaging systems.

Project Status

Deckard has been used in a production environment since 2019, handling billions of messages and thousands of requests per second. Since its initial internal release, Deckard has undergone significant improvements and enhancements.

The performance and reliability of Deckard are directly dependent on the storage and cache services. Therefore, it is crucial to configure MongoDB and Redis correctly to ensure optimal performance. Redis is responsible for storing all priority queues and has a significant impact on Deckard's performance, while MongoDB usage is simpler and has a lower impact on performance.

Currently, we have projects that utilize MongoDB in both a virtual machine environment and a Kubernetes environment. However, it's worth noting that we have had more extensive usage of MongoDB in a virtual machine (VM) environment compared to Kubernetes. When deploying Deckard, it is important to configure MongoDB properly, following MongoDB's recommendations for production environments (documented here: MongoDB Production Notes).

While the project is being released with a 0.0.x version due to necessary code modifications for open-sourcing, it has been thoroughly tested and proven to be reliable in production scenarios.

Please refer to our issues section for more details. Your feedback and suggestions are highly appreciated and can be shared in our discussions.

Getting Started

To quickly get started with Deckard, please consult our getting started guide, which provides step-by-step instructions. Additionally, we have provided client documentation for various programming languages to assist you in integrating Deckard into your projects.

You can also access our documentation on Deckard's components here. More details on the project structure can be found here.

Running Deckard

Here's a quick guide on how to run Deckard. You should check the getting started guide for more details.

On Linux you can run it from sources with:

make run

You can run it with Docker:

docker run --rm -p 8081:8081 blipai/deckard

By default for Docker and Linux it will use a memory storage and a memory cache engine.

To change the default configuration see the configuration section.

You can also run it in a Kubernetes cluster using Helm:

It will deploy a MongoDB for storage and a Redis for cache.

Check the chart values.yaml to see all available configurations.

helm repo add deckard https://takenet.github.io/deckard/
helm install deckard deckard/deckard

You may also download the latest release from the releases page and execute it.

Configuration

We currently use the viper project to manage configurations and the current implementation delegates the configuration to environment variables.

All available environment variables are listed below:

Overall Configuration
Environment Variable Default Description
DECKARD_DEBUG false To enable debug mode to log more information.
DECKARD_LOG_TYPE json The log type to use. Available: json, text
DECKARD_GRPC_ENABLED true To enable the gRPC service. You can disable gRPC service if you want an instance to perform only housekeeper tasks.
DECKARD_GRPC_PORT 8081 The gRPC port to listen.
Cache Configuration
Environment Variable Default Description
DECKARD_CACHE_TYPE MEMORY The cache implementation to use. Available: MEMORY, REDIS
DECKARD_CACHE_URI The cache Connection URI to connect with the cache service. Currently only a Redis URI is accepted. It will take precedence over any other environment variable related to the connection Redis connection.
DECKARD_REDIS_ADDRESS localhost The redis address to connect while using redis cache implementation. It will be overriden by DECKARD_CACHE_URI if present.
DECKARD_REDIS_PASSWORD The redis password to connect while using redis cache implementation. It will be overriden by DECKARD_CACHE_URI if present.
DECKARD_REDIS_PORT 6379 The redis port to connect while using redis cache implementation. It will be overriden by DECKARD_CACHE_URI if present.
DECKARD_REDIS_DB 0 The database to use while using redis cache implementation. It will be overriden by DECKARD_CACHE_URI if present.
Storage Configuration
Environment Variable Default Description
DECKARD_STORAGE_TYPE MEMORY The storage implementation to use. Available: MEMORY, MONGODB
DECKARD_STORAGE_URI The storage Connection URI to connect with the storage service. Currently only a MongoDB URI is accepted. It can override any other environment variable related to the connection MongoDB connection since it takes precedence.
DECKARD_MONGODB_ADDRESSES localhost:27017 The MongoDB addresses separated by comma to connect while using MongoDB storage implementation. It can be overridden by DECKARD_STORAGE_URI.
DECKARD_MONGODB_AUTH_DB The MongoDB auth database to authenticate while using MongoDB storage implementation. It can be overridden by DECKARD_STORAGE_URI.
DECKARD_MONGODB_PASSWORD The MongoDB password to authenticate while using MongoDB storage implementation. It can be overridden by DECKARD_STORAGE_URI.
DECKARD_MONGODB_DATABASE deckard The MongoDB database to use to store messages while using MongoDB storage implementation.
DECKARD_MONGODB_COLLECTION queue The MongoDB collection to use to store messages while using MongoDB storage implementation.
DECKARD_MONGODB_USER The MongoDB user to authenticate while using MongoDB storage implementation. It can be overridden by DECKARD_STORAGE_URI.
DECKARD_MONGODB_SSL false To enable SSL while using MongoDB storage implementation. It can be overridden by DECKARD_STORAGE_URI.
DECKARD_MONGODB_QUEUE_CONFIGURATION_COLLECTION queue_configuration The MongoDB collection to use to store queue configurations while using MongoDB storage implementation.
Housekeeper Configuration
Environment Variable Default Description
DECKARD_HOUSEKEEPER_ENABLED true To enable housekeeper tasks.
DECKARD_HOUSEKEEPER_TASK_TIMEOUT_DELAY 1s The delay between each timeout task execution.
DECKARD_HOUSEKEEPER_TASK_UNLOCK_DELAY 1s The delay between each unlock task execution.
DECKARD_HOUSEKEEPER_TASK_UPDATE_DELAY 1s The delay between each update task execution.
DECKARD_HOUSEKEEPER_TASK_TTL_DELAY 1s The delay between each ttl task execution.
DECKARD_HOUSEKEEPER_TASK_MAX_ELEMENTS_DELAY 1s The delay between each max elements task execution.
DECKARD_HOUSEKEEPER_TASK_METRICS_DELAY 60s The delay between each metrics task execution.
Audit Configuration
Environment Variable Default Description
DECKARD_AUDIT_ENABLED false To enable auditing.
DECKARD_ELASTIC_ADDRESS http://localhost:9200/ A ElasticSearch address to connect to store audit information.
DECKARD_ELASTIC_PASSWORD A ElasticSearch password to connect to store audit information.
DECKARD_ELASTIC_USER A ElasticSearch user to connect to store audit information.
Metrics Configuration
Environment Variable Default Description
DECKARD_METRICS_ENABLED true To enable exporting metrics to the metrics endpoint.
DECKARD_METRICS_PORT 22022 The metrics http port to listn.
DECKARD_METRICS_PATH /metrics The metrics http path to expose metrics.
DECKARD_METRICS_HISTOGRAM_BUCKETS 0,1,2,5,10,15,20,30,35,50,100,200,400,600,800,1000,1500,2000,5000,10000 The histogram buckets to use to expose metrics.
DECKARD_METRICS_OPENMETRICS_ENABLED true If true, the OpenMetrics encoding is added to the possible options during content negotiation.
TLS Configuration

To learn more about gRPC TLS configuration please refer to gRPC Auth.

Environment Variable Default Description
DECKARD_TLS_CLIENT_AUTH_TYPE NoClientCert The type of client authentication TLS verification to use. Available: NoClientCert, RequestClientCert, RequireAnyClientCert, VerifyClientCertIfGiven, RequireAndVerifyClientCert.
DECKARD_TLS_SERVER_CERT_FILE_PATHS A comma-delimited list of absolute file paths to PEM-encoded certificates.
DECKARD_TLS_SERVER_KEY_FILE_PATHS A comma-delimited list of absolute file paths to PEM-encoded private keys.
DECKARD_TLS_CLIENT_CERT_FILE_PATHS A comma-delimited list of absolute file paths to PEM-encoded certificates to enable mutual TLS.

Contributing

We are always looking for new contributors to help us improve Deckard.

If you want to contribute to Deckard, please read our contributing guide which includes how to build, run and test Deckard and a complete description of our project structure.

License

Deckard is licensed under the MIT License.

Acknowledgments

We would like to thank the following people for their initial contributions building Deckard's first version:

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var Deckard_ServiceDesc = grpc.ServiceDesc{
	ServiceName: "blipai.deckard.Deckard",
	HandlerType: (*DeckardServer)(nil),
	Methods: []grpc.MethodDesc{
		{
			MethodName: "Add",
			Handler:    _Deckard_Add_Handler,
		},
		{
			MethodName: "Pull",
			Handler:    _Deckard_Pull_Handler,
		},
		{
			MethodName: "Ack",
			Handler:    _Deckard_Ack_Handler,
		},
		{
			MethodName: "Nack",
			Handler:    _Deckard_Nack_Handler,
		},
		{
			MethodName: "Count",
			Handler:    _Deckard_Count_Handler,
		},
		{
			MethodName: "Remove",
			Handler:    _Deckard_Remove_Handler,
		},
		{
			MethodName: "Flush",
			Handler:    _Deckard_Flush_Handler,
		},
		{
			MethodName: "GetById",
			Handler:    _Deckard_GetById_Handler,
		},
		{
			MethodName: "EditQueue",
			Handler:    _Deckard_EditQueue_Handler,
		},
		{
			MethodName: "GetQueue",
			Handler:    _Deckard_GetQueue_Handler,
		},
	},
	Streams:  []grpc.StreamDesc{},
	Metadata: "deckard_service.proto",
}

Deckard_ServiceDesc is the grpc.ServiceDesc for Deckard service. It's only intended for direct use with grpc.RegisterService, and not to be introspected or modified (even as a copy)

View Source 
var File_deckard_service_proto protoreflect.FileDescriptor

Functions

func RegisterDeckardServer 

func RegisterDeckardServer(s grpc.ServiceRegistrar, srv DeckardServer)

Types

type AckRequest 

type AckRequest struct {

	// ID of the message
	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
	// Queue where this message is stored
	Queue string `protobuf:"bytes,2,opt,name=queue,proto3" json:"queue,omitempty"`
	// Reason of this result.
	// Useful for audit, mostly on 'nack' signals.
	Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"`
	// The value to subtract the score and increase final message score.
	// For example if you want to make this message to have a better score you can add 10000 which will represent 10s of score benefit.
	// If you want to penalize the message you can send a negative number.
	//
	// IMPORTANT: The message will not be locked by, in the example, 10 seconds. This attribute is used only to increase or decrease the message priority in the priority queue.
	//
	// This attribute is used only for ack requests and can't be used at the same time of 'lock_ms' attribute.
	ScoreSubtract float64 `protobuf:"fixed64,3,opt,name=score_subtract,json=scoreSubtract,proto3" json:"score_subtract,omitempty"`
	// Breakpoint to set for this message
	Breakpoint string `protobuf:"bytes,4,opt,name=breakpoint,proto3" json:"breakpoint,omitempty"`
	// Time in milliseconds to lock a message before returning it to the queue.
	// For nack requests the message will be locked before returning to first position in the priority queue.
	// For ack requests the message will be locked before returning to last position in the priority queue.
	//
	// IMPORTANT: The 'score_subtract' attribute will be ignored if this attribute is different than 0.
	//
	// IMPORTANT: Deckard checks for locked messages in a 1-second delay meaning the lock have a second precision and not milliseconds.
	// This field is in milliseconds because all scores and duration units on deckard are expressed in milliseconds.
	LockMs int64 `protobuf:"varint,6,opt,name=lock_ms,json=lockMs,proto3" json:"lock_ms,omitempty"`
	// Whether the message should be removed when acked/nacked
	RemoveMessage bool `protobuf:"varint,7,opt,name=removeMessage,proto3" json:"removeMessage,omitempty"`
	// contains filtered or unexported fields
}

func (*AckRequest) Descriptor deprecated 

func (*AckRequest) Descriptor() ([]byte, []int)

Deprecated: Use AckRequest.ProtoReflect.Descriptor instead.

func (*AckRequest) GetBreakpoint 

func (x *AckRequest) GetBreakpoint() string

func (*AckRequest) GetId 

func (x *AckRequest) GetId() string

func (*AckRequest) GetLockMs 

func (x *AckRequest) GetLockMs() int64

func (*AckRequest) GetQueue 

func (x *AckRequest) GetQueue() string

func (*AckRequest) GetReason 

func (x *AckRequest) GetReason() string

func (*AckRequest) GetRemoveMessage 

func (x *AckRequest) GetRemoveMessage() bool

func (*AckRequest) GetScoreSubtract 

func (x *AckRequest) GetScoreSubtract() float64

func (*AckRequest) ProtoMessage 

func (*AckRequest) ProtoMessage()

func (*AckRequest) ProtoReflect 

func (x *AckRequest) ProtoReflect() protoreflect.Message

func (*AckRequest) Reset 

func (x *AckRequest) Reset()

func (*AckRequest) String 

func (x *AckRequest) String() string

type AckResponse 

type AckResponse struct {
	Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"`
	// The removal response if the message's removal was requested
	RemovalResponse *RemoveResponse `protobuf:"bytes,2,opt,name=removal_response,json=removalResponse,proto3" json:"removal_response,omitempty"`
	// contains filtered or unexported fields
}

func (*AckResponse) Descriptor deprecated 

func (*AckResponse) Descriptor() ([]byte, []int)

Deprecated: Use AckResponse.ProtoReflect.Descriptor instead.

func (*AckResponse) GetRemovalResponse 

func (x *AckResponse) GetRemovalResponse() *RemoveResponse

func (*AckResponse) GetSuccess 

func (x *AckResponse) GetSuccess() bool

func (*AckResponse) ProtoMessage 

func (*AckResponse) ProtoMessage()

func (*AckResponse) ProtoReflect 

func (x *AckResponse) ProtoReflect() protoreflect.Message

func (*AckResponse) Reset 

func (x *AckResponse) Reset()

func (*AckResponse) String 

func (x *AckResponse) String() string

type AddMessage 

type AddMessage struct {

	// Unique id of this message
	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
	// A payload map with formatted data to be stored and used by clients.
	Payload map[string]*anypb.Any `` /* 156-byte string literal not displayed */
	// Non-formatted string payload.
	// This field can be used to store simple string data instead of using the payload field.
	StringPayload string `protobuf:"bytes,3,opt,name=string_payload,json=stringPayload,proto3" json:"string_payload,omitempty"`
	// Metadata is a map of string to be used as a key-value store.
	// It is a simple way to store data that is not part of the message payload.
	Metadata map[string]string `` /* 158-byte string literal not displayed */
	//
	//Name of the queue to add this message
	//
	//Suggestion: to better observability, provide the name of the application using colon as the separator. Example: <application_name>:<queue_name>
	//
	//You may also add a queue prefix to the queue name using two colons as the separator. Example: <application_name>:<queue_prefix>::<queue_name>
	Queue string `protobuf:"bytes,4,opt,name=queue,proto3" json:"queue,omitempty"`
	// Indicate this message will never expire and will only be deleted from the queue if explicitly removed.
	Timeless bool `protobuf:"varint,6,opt,name=timeless,proto3" json:"timeless,omitempty"`
	// TTL is the amount of time in minutes this message will live in the queue.
	// TTL is not a exact time, the message may live a little longer than the specified TTL.
	TtlMinutes int64 `protobuf:"varint,7,opt,name=ttl_minutes,json=ttlMinutes,proto3" json:"ttl_minutes,omitempty"`
	// Description of the message, this should be used as a human readable string to be used in diagnostics.
	Description string `protobuf:"bytes,8,opt,name=description,proto3" json:"description,omitempty"`
	// contains filtered or unexported fields
}

func (*AddMessage) Descriptor deprecated 

func (*AddMessage) Descriptor() ([]byte, []int)

Deprecated: Use AddMessage.ProtoReflect.Descriptor instead.

func (*AddMessage) GetDescription 

func (x *AddMessage) GetDescription() string

func (*AddMessage) GetId 

func (x *AddMessage) GetId() string

func (*AddMessage) GetMetadata 

func (x *AddMessage) GetMetadata() map[string]string

func (*AddMessage) GetPayload 

func (x *AddMessage) GetPayload() map[string]*anypb.Any

func (*AddMessage) GetQueue 

func (x *AddMessage) GetQueue() string

func (*AddMessage) GetStringPayload 

func (x *AddMessage) GetStringPayload() string

func (*AddMessage) GetTimeless 

func (x *AddMessage) GetTimeless() bool

func (*AddMessage) GetTtlMinutes 

func (x *AddMessage) GetTtlMinutes() int64

func (*AddMessage) ProtoMessage 

func (*AddMessage) ProtoMessage()

func (*AddMessage) ProtoReflect 

func (x *AddMessage) ProtoReflect() protoreflect.Message

func (*AddMessage) Reset 

func (x *AddMessage) Reset()

func (*AddMessage) String 

func (x *AddMessage) String() string

type AddRequest 

type AddRequest struct {

	// List of messages to be added
	Messages []*AddMessage `protobuf:"bytes,1,rep,name=messages,proto3" json:"messages,omitempty"`
	// contains filtered or unexported fields
}

func (*AddRequest) Descriptor deprecated 

func (*AddRequest) Descriptor() ([]byte, []int)

Deprecated: Use AddRequest.ProtoReflect.Descriptor instead.

func (*AddRequest) GetMessages 

func (x *AddRequest) GetMessages() []*AddMessage

func (*AddRequest) ProtoMessage 

func (*AddRequest) ProtoMessage()

func (*AddRequest) ProtoReflect 

func (x *AddRequest) ProtoReflect() protoreflect.Message

func (*AddRequest) Reset 

func (x *AddRequest) Reset()

func (*AddRequest) String 

func (x *AddRequest) String() string

type AddResponse 

type AddResponse struct {

	// Number of created messages
	CreatedCount int64 `protobuf:"varint,1,opt,name=created_count,json=createdCount,proto3" json:"created_count,omitempty"`
	// Number of updated messages
	UpdatedCount int64 `protobuf:"varint,2,opt,name=updated_count,json=updatedCount,proto3" json:"updated_count,omitempty"`
	// contains filtered or unexported fields
}

func (*AddResponse) Descriptor deprecated 

func (*AddResponse) Descriptor() ([]byte, []int)

Deprecated: Use AddResponse.ProtoReflect.Descriptor instead.

func (*AddResponse) GetCreatedCount 

func (x *AddResponse) GetCreatedCount() int64

func (*AddResponse) GetUpdatedCount 

func (x *AddResponse) GetUpdatedCount() int64

func (*AddResponse) ProtoMessage 

func (*AddResponse) ProtoMessage()

func (*AddResponse) ProtoReflect 

func (x *AddResponse) ProtoReflect() protoreflect.Message

func (*AddResponse) Reset 

func (x *AddResponse) Reset()

func (*AddResponse) String 

func (x *AddResponse) String() string

type CountRequest 

type CountRequest struct {
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// contains filtered or unexported fields
}

func (*CountRequest) Descriptor deprecated 

func (*CountRequest) Descriptor() ([]byte, []int)

Deprecated: Use CountRequest.ProtoReflect.Descriptor instead.

func (*CountRequest) GetQueue 

func (x *CountRequest) GetQueue() string

func (*CountRequest) ProtoMessage 

func (*CountRequest) ProtoMessage()

func (*CountRequest) ProtoReflect 

func (x *CountRequest) ProtoReflect() protoreflect.Message

func (*CountRequest) Reset 

func (x *CountRequest) Reset()

func (*CountRequest) String 

func (x *CountRequest) String() string

type CountResponse 

type CountResponse struct {
	Count int64 `protobuf:"varint,1,opt,name=count,proto3" json:"count,omitempty"`
	// contains filtered or unexported fields
}

func (*CountResponse) Descriptor deprecated 

func (*CountResponse) Descriptor() ([]byte, []int)

Deprecated: Use CountResponse.ProtoReflect.Descriptor instead.

func (*CountResponse) GetCount 

func (x *CountResponse) GetCount() int64

func (*CountResponse) ProtoMessage 

func (*CountResponse) ProtoMessage()

func (*CountResponse) ProtoReflect 

func (x *CountResponse) ProtoReflect() protoreflect.Message

func (*CountResponse) Reset 

func (x *CountResponse) Reset()

func (*CountResponse) String 

func (x *CountResponse) String() string

type DeckardClient 

type DeckardClient interface {
	// Adds messages to the queue. If any message already exists (same id and queue) it will be updated
	Add(ctx context.Context, in *AddRequest, opts ...grpc.CallOption) (*AddResponse, error)
	// Pulls messages from the queue
	Pull(ctx context.Context, in *PullRequest, opts ...grpc.CallOption) (*PullResponse, error)
	// Acks a message, marking it as successfully processed
	Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error)
	// Nacks a message, marking it as not processed
	Nack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error)
	// Counts number of messages on queue based on a filter request
	Count(ctx context.Context, in *CountRequest, opts ...grpc.CallOption) (*CountResponse, error)
	// Removes one or many messages from the queue based on its ids
	Remove(ctx context.Context, in *RemoveRequest, opts ...grpc.CallOption) (*RemoveResponse, error)
	// Discards all deckard data from storage and cache.
	// This request is only available if deckard instance is configured with MEMORY cache and storage.
	Flush(ctx context.Context, in *FlushRequest, opts ...grpc.CallOption) (*FlushResponse, error)
	// Gets a message from a specific queue using its id
	// CAUTION: this should be used mainly for diagnostics and debugging purposes since it will be direct operation on the storage system
	GetById(ctx context.Context, in *GetByIdRequest, opts ...grpc.CallOption) (*GetByIdResponse, error)
	// Edits a queue configuration
	EditQueue(ctx context.Context, in *EditQueueRequest, opts ...grpc.CallOption) (*EditQueueResponse, error)
	// Gets the current configuration for a queue
	GetQueue(ctx context.Context, in *GetQueueRequest, opts ...grpc.CallOption) (*GetQueueResponse, error)
}

DeckardClient is the client API for Deckard service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.

func NewDeckardClient 

func NewDeckardClient(cc grpc.ClientConnInterface) DeckardClient

type DeckardServer 

type DeckardServer interface {
	// Adds messages to the queue. If any message already exists (same id and queue) it will be updated
	Add(context.Context, *AddRequest) (*AddResponse, error)
	// Pulls messages from the queue
	Pull(context.Context, *PullRequest) (*PullResponse, error)
	// Acks a message, marking it as successfully processed
	Ack(context.Context, *AckRequest) (*AckResponse, error)
	// Nacks a message, marking it as not processed
	Nack(context.Context, *AckRequest) (*AckResponse, error)
	// Counts number of messages on queue based on a filter request
	Count(context.Context, *CountRequest) (*CountResponse, error)
	// Removes one or many messages from the queue based on its ids
	Remove(context.Context, *RemoveRequest) (*RemoveResponse, error)
	// Discards all deckard data from storage and cache.
	// This request is only available if deckard instance is configured with MEMORY cache and storage.
	Flush(context.Context, *FlushRequest) (*FlushResponse, error)
	// Gets a message from a specific queue using its id
	// CAUTION: this should be used mainly for diagnostics and debugging purposes since it will be direct operation on the storage system
	GetById(context.Context, *GetByIdRequest) (*GetByIdResponse, error)
	// Edits a queue configuration
	EditQueue(context.Context, *EditQueueRequest) (*EditQueueResponse, error)
	// Gets the current configuration for a queue
	GetQueue(context.Context, *GetQueueRequest) (*GetQueueResponse, error)
	// contains filtered or unexported methods
}

DeckardServer is the server API for Deckard service. All implementations must embed UnimplementedDeckardServer for forward compatibility

type EditQueueRequest 

type EditQueueRequest struct {

	// Name of the queue to be updated
	// This includes all prefixes and suffixes
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// Configuration to apply to the queue. It will always update the queue with the newer configuration.
	// Only available fields will be updated, meaning that previously configured attributes will not be change unless you explicit set it.
	// If you want to change a configuration to its default value, manually set it to its default value following each field documentation.
	Configuration *QueueConfiguration `protobuf:"bytes,2,opt,name=configuration,proto3" json:"configuration,omitempty"`
	// contains filtered or unexported fields
}

func (*EditQueueRequest) Descriptor deprecated 

func (*EditQueueRequest) Descriptor() ([]byte, []int)

Deprecated: Use EditQueueRequest.ProtoReflect.Descriptor instead.

func (*EditQueueRequest) GetConfiguration 

func (x *EditQueueRequest) GetConfiguration() *QueueConfiguration

func (*EditQueueRequest) GetQueue 

func (x *EditQueueRequest) GetQueue() string

func (*EditQueueRequest) ProtoMessage 

func (*EditQueueRequest) ProtoMessage()

func (*EditQueueRequest) ProtoReflect 

func (x *EditQueueRequest) ProtoReflect() protoreflect.Message

func (*EditQueueRequest) Reset 

func (x *EditQueueRequest) Reset()

func (*EditQueueRequest) String 

func (x *EditQueueRequest) String() string

type EditQueueResponse 

type EditQueueResponse struct {

	// Name of the queue
	Queue   string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	Success bool   `protobuf:"varint,2,opt,name=success,proto3" json:"success,omitempty"`
	// contains filtered or unexported fields
}

func (*EditQueueResponse) Descriptor deprecated 

func (*EditQueueResponse) Descriptor() ([]byte, []int)

Deprecated: Use EditQueueResponse.ProtoReflect.Descriptor instead.

func (*EditQueueResponse) GetQueue 

func (x *EditQueueResponse) GetQueue() string

func (*EditQueueResponse) GetSuccess 

func (x *EditQueueResponse) GetSuccess() bool

func (*EditQueueResponse) ProtoMessage 

func (*EditQueueResponse) ProtoMessage()

func (*EditQueueResponse) ProtoReflect 

func (x *EditQueueResponse) ProtoReflect() protoreflect.Message

func (*EditQueueResponse) Reset 

func (x *EditQueueResponse) Reset()

func (*EditQueueResponse) String 

func (x *EditQueueResponse) String() string

type FlushRequest 

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

func (*FlushRequest) Descriptor deprecated 

func (*FlushRequest) Descriptor() ([]byte, []int)

Deprecated: Use FlushRequest.ProtoReflect.Descriptor instead.

func (*FlushRequest) ProtoMessage 

func (*FlushRequest) ProtoMessage()

func (*FlushRequest) ProtoReflect 

func (x *FlushRequest) ProtoReflect() protoreflect.Message

func (*FlushRequest) Reset 

func (x *FlushRequest) Reset()

func (*FlushRequest) String 

func (x *FlushRequest) String() string

type FlushResponse 

type FlushResponse struct {
	Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"`
	// contains filtered or unexported fields
}

func (*FlushResponse) Descriptor deprecated 

func (*FlushResponse) Descriptor() ([]byte, []int)

Deprecated: Use FlushResponse.ProtoReflect.Descriptor instead.

func (*FlushResponse) GetSuccess 

func (x *FlushResponse) GetSuccess() bool

func (*FlushResponse) ProtoMessage 

func (*FlushResponse) ProtoMessage()

func (*FlushResponse) ProtoReflect 

func (x *FlushResponse) ProtoReflect() protoreflect.Message

func (*FlushResponse) Reset 

func (x *FlushResponse) Reset()

func (*FlushResponse) String 

func (x *FlushResponse) String() string

type GetByIdRequest 

type GetByIdRequest struct {

	// Name of the queue to get a message
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// The message id
	Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"`
	// contains filtered or unexported fields
}

func (*GetByIdRequest) Descriptor deprecated 

func (*GetByIdRequest) Descriptor() ([]byte, []int)

Deprecated: Use GetByIdRequest.ProtoReflect.Descriptor instead.

func (*GetByIdRequest) GetId 

func (x *GetByIdRequest) GetId() string

func (*GetByIdRequest) GetQueue 

func (x *GetByIdRequest) GetQueue() string

func (*GetByIdRequest) ProtoMessage 

func (*GetByIdRequest) ProtoMessage()

func (*GetByIdRequest) ProtoReflect 

func (x *GetByIdRequest) ProtoReflect() protoreflect.Message

func (*GetByIdRequest) Reset 

func (x *GetByIdRequest) Reset()

func (*GetByIdRequest) String 

func (x *GetByIdRequest) String() string

type GetByIdResponse 

type GetByIdResponse struct {
	Message *Message `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
	// A human readable string data map when the message is found
	// This represents the payload map as a JSON string representation
	// This is useful for debugging and diagnostics
	HumanReadablePayload map[string]string `` /* 211-byte string literal not displayed */
	Found                bool              `protobuf:"varint,2,opt,name=found,proto3" json:"found,omitempty"`
	// contains filtered or unexported fields
}

func (*GetByIdResponse) Descriptor deprecated 

func (*GetByIdResponse) Descriptor() ([]byte, []int)

Deprecated: Use GetByIdResponse.ProtoReflect.Descriptor instead.

func (*GetByIdResponse) GetFound 

func (x *GetByIdResponse) GetFound() bool

func (*GetByIdResponse) GetHumanReadablePayload 

func (x *GetByIdResponse) GetHumanReadablePayload() map[string]string

func (*GetByIdResponse) GetMessage 

func (x *GetByIdResponse) GetMessage() *Message

func (*GetByIdResponse) ProtoMessage 

func (*GetByIdResponse) ProtoMessage()

func (*GetByIdResponse) ProtoReflect 

func (x *GetByIdResponse) ProtoReflect() protoreflect.Message

func (*GetByIdResponse) Reset 

func (x *GetByIdResponse) Reset()

func (*GetByIdResponse) String 

func (x *GetByIdResponse) String() string

type GetQueueRequest 

type GetQueueRequest struct {

	// Name of the queue to be updated
	// This includes all prefixes and suffixes
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// contains filtered or unexported fields
}

func (*GetQueueRequest) Descriptor deprecated 

func (*GetQueueRequest) Descriptor() ([]byte, []int)

Deprecated: Use GetQueueRequest.ProtoReflect.Descriptor instead.

func (*GetQueueRequest) GetQueue 

func (x *GetQueueRequest) GetQueue() string

func (*GetQueueRequest) ProtoMessage 

func (*GetQueueRequest) ProtoMessage()

func (*GetQueueRequest) ProtoReflect 

func (x *GetQueueRequest) ProtoReflect() protoreflect.Message

func (*GetQueueRequest) Reset 

func (x *GetQueueRequest) Reset()

func (*GetQueueRequest) String 

func (x *GetQueueRequest) String() string

type GetQueueResponse 

type GetQueueResponse struct {

	// Name of the queue
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// Configuration of the queue
	Configuration *QueueConfiguration `protobuf:"bytes,2,opt,name=configuration,proto3" json:"configuration,omitempty"`
	// contains filtered or unexported fields
}

func (*GetQueueResponse) Descriptor deprecated 

func (*GetQueueResponse) Descriptor() ([]byte, []int)

Deprecated: Use GetQueueResponse.ProtoReflect.Descriptor instead.

func (*GetQueueResponse) GetConfiguration 

func (x *GetQueueResponse) GetConfiguration() *QueueConfiguration

func (*GetQueueResponse) GetQueue 

func (x *GetQueueResponse) GetQueue() string

func (*GetQueueResponse) ProtoMessage 

func (*GetQueueResponse) ProtoMessage()

func (*GetQueueResponse) ProtoReflect 

func (x *GetQueueResponse) ProtoReflect() protoreflect.Message

func (*GetQueueResponse) Reset 

func (x *GetQueueResponse) Reset()

func (*GetQueueResponse) String 

func (x *GetQueueResponse) String() string

type Message 

type Message struct {

	// ID is an unique identifier inside a queue.
	// Any message with the same id and queue will be considered the same message.
	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
	// Description of the message, this should be used as a human readable string to be used in diagnostics.
	Description string `protobuf:"bytes,2,opt,name=description,proto3" json:"description,omitempty"`
	// Full name of the queue this message belongs (including any prefixes)
	Queue string `protobuf:"bytes,3,opt,name=queue,proto3" json:"queue,omitempty"`
	// A payload map with formatted data to be stored and used by clients.
	Payload map[string]*anypb.Any `` /* 155-byte string literal not displayed */
	// Metadata is a map of string to be used as a key-value store.
	// It is a simple way to store data that is not part of the message payload.
	Metadata map[string]string `` /* 157-byte string literal not displayed */
	// Message string payload. Is responsibility of the caller to know how to encode/decode to a useful format for its purpose.
	// This field can be used to store simple string data instead of using the payload field.
	StringPayload string `protobuf:"bytes,5,opt,name=string_payload,json=stringPayload,proto3" json:"string_payload,omitempty"`
	// Score is the priority this message currently have in the queue.
	Score float64 `protobuf:"fixed64,6,opt,name=score,proto3" json:"score,omitempty"`
	// Breakpoint is a field to be used as an auxiliar field for some specific use cases.
	//
	// For example if you need to keep a record of the last result processing a message, use this field like iteracting with a pagination system.
	Breakpoint string `protobuf:"bytes,7,opt,name=breakpoint,proto3" json:"breakpoint,omitempty"`
	// contains filtered or unexported fields
}

func (*Message) Descriptor deprecated 

func (*Message) Descriptor() ([]byte, []int)

Deprecated: Use Message.ProtoReflect.Descriptor instead.

func (*Message) GetBreakpoint 

func (x *Message) GetBreakpoint() string

func (*Message) GetDescription 

func (x *Message) GetDescription() string

func (*Message) GetId 

func (x *Message) GetId() string

func (*Message) GetMetadata 

func (x *Message) GetMetadata() map[string]string

func (*Message) GetPayload 

func (x *Message) GetPayload() map[string]*anypb.Any

func (*Message) GetQueue 

func (x *Message) GetQueue() string

func (*Message) GetScore 

func (x *Message) GetScore() float64

func (*Message) GetStringPayload 

func (x *Message) GetStringPayload() string

func (*Message) ProtoMessage 

func (*Message) ProtoMessage()

func (*Message) ProtoReflect 

func (x *Message) ProtoReflect() protoreflect.Message

func (*Message) Reset 

func (x *Message) Reset()

func (*Message) String 

func (x *Message) String() string

type PullRequest 

type PullRequest struct {

	// Full name of the queue to pull messages (including any prefixes)
	Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"`
	// Number of messages to fetch.
	// Caution: as greater the amount, as more time it will take to process the request.
	// Max value is 1000 and the default value is 1
	Amount int32 `protobuf:"varint,2,opt,name=amount,proto3" json:"amount,omitempty"`
	//
	//Number to subtract to the current time to filter the max score to return.
	//Useful to not return a message just moments after it was last used.
	//
	//For example if in your queue the score is only based on the time (always acking with score_subtract as 0),
	//this parameter will be the number of milliseconds since the message's last usage.
	ScoreFilter int64 `protobuf:"varint,3,opt,name=score_filter,json=scoreFilter,proto3" json:"score_filter,omitempty"`
	// contains filtered or unexported fields
}

func (*PullRequest) Descriptor deprecated 

func (*PullRequest) Descriptor() ([]byte, []int)

Deprecated: Use PullRequest.ProtoReflect.Descriptor instead.

func (*PullRequest) GetAmount 

func (x *PullRequest) GetAmount() int32

func (*PullRequest) GetQueue 

func (x *PullRequest) GetQueue() string

func (*PullRequest) GetScoreFilter 

func (x *PullRequest) GetScoreFilter() int64

func (*PullRequest) ProtoMessage 

func (*PullRequest) ProtoMessage()

func (*PullRequest) ProtoReflect 

func (x *PullRequest) ProtoReflect() protoreflect.Message

func (*PullRequest) Reset 

func (x *PullRequest) Reset()

func (*PullRequest) String 

func (x *PullRequest) String() string

type PullResponse 

type PullResponse struct {

	// List of returned messages
	Messages []*Message `protobuf:"bytes,1,rep,name=messages,proto3" json:"messages,omitempty"`
	// contains filtered or unexported fields
}

func (*PullResponse) Descriptor deprecated 

func (*PullResponse) Descriptor() ([]byte, []int)

Deprecated: Use PullResponse.ProtoReflect.Descriptor instead.

func (*PullResponse) GetMessages 

func (x *PullResponse) GetMessages() []*Message

func (*PullResponse) ProtoMessage 

func (*PullResponse) ProtoMessage()

func (*PullResponse) ProtoReflect 

func (x *PullResponse) ProtoReflect() protoreflect.Message

func (*PullResponse) Reset 

func (x *PullResponse) Reset()

func (*PullResponse) String 

func (x *PullResponse) String() string

type QueueConfiguration 

type QueueConfiguration struct {

	//
	//Number of max elements the queue can have.
	//
	//To apply a max elements to a queue, set a value greater than 0.
	//To remove the max elements from a queue, set the value to -1.
	//0 will be always ignored and the queue will not be updated.
	//
	//All queues are unlimited by default.
	MaxElements int64 `protobuf:"varint,1,opt,name=max_elements,json=maxElements,proto3" json:"max_elements,omitempty"`
	// contains filtered or unexported fields
}

The queue configuration does not change instantly and can take up to 10 minutes to complete update.

func (*QueueConfiguration) Descriptor deprecated 

func (*QueueConfiguration) Descriptor() ([]byte, []int)

Deprecated: Use QueueConfiguration.ProtoReflect.Descriptor instead.

func (*QueueConfiguration) GetMaxElements 

func (x *QueueConfiguration) GetMaxElements() int64

func (*QueueConfiguration) ProtoMessage 

func (*QueueConfiguration) ProtoMessage()

func (*QueueConfiguration) ProtoReflect 

func (x *QueueConfiguration) ProtoReflect() protoreflect.Message

func (*QueueConfiguration) Reset 

func (x *QueueConfiguration) Reset()

func (*QueueConfiguration) String 

func (x *QueueConfiguration) String() string

type RemoveRequest 

type RemoveRequest struct {
	Ids []string `protobuf:"bytes,1,rep,name=ids,proto3" json:"ids,omitempty"`
	// Name of the queue to remove message
	// Provide the name of the application as a prefix using colon as the separator. Example: <application_name>:<queue_name>
	Queue string `protobuf:"bytes,2,opt,name=queue,proto3" json:"queue,omitempty"`
	// contains filtered or unexported fields
}

func (*RemoveRequest) Descriptor deprecated 

func (*RemoveRequest) Descriptor() ([]byte, []int)

Deprecated: Use RemoveRequest.ProtoReflect.Descriptor instead.

func (*RemoveRequest) GetIds 

func (x *RemoveRequest) GetIds() []string

func (*RemoveRequest) GetQueue 

func (x *RemoveRequest) GetQueue() string

func (*RemoveRequest) ProtoMessage 

func (*RemoveRequest) ProtoMessage()

func (*RemoveRequest) ProtoReflect 

func (x *RemoveRequest) ProtoReflect() protoreflect.Message

func (*RemoveRequest) Reset 

func (x *RemoveRequest) Reset()

func (*RemoveRequest) String 

func (x *RemoveRequest) String() string

type RemoveResponse 

type RemoveResponse struct {
	CacheRemoved   int64 `protobuf:"varint,1,opt,name=cacheRemoved,proto3" json:"cacheRemoved,omitempty"`
	StorageRemoved int64 `protobuf:"varint,2,opt,name=storageRemoved,proto3" json:"storageRemoved,omitempty"`
	// contains filtered or unexported fields
}

func (*RemoveResponse) Descriptor deprecated 

func (*RemoveResponse) Descriptor() ([]byte, []int)

Deprecated: Use RemoveResponse.ProtoReflect.Descriptor instead.

func (*RemoveResponse) GetCacheRemoved 

func (x *RemoveResponse) GetCacheRemoved() int64

func (*RemoveResponse) GetStorageRemoved 

func (x *RemoveResponse) GetStorageRemoved() int64

func (*RemoveResponse) ProtoMessage 

func (*RemoveResponse) ProtoMessage()

func (*RemoveResponse) ProtoReflect 

func (x *RemoveResponse) ProtoReflect() protoreflect.Message

func (*RemoveResponse) Reset 

func (x *RemoveResponse) Reset()

func (*RemoveResponse) String 

func (x *RemoveResponse) String() string

type UnimplementedDeckardServer 

type UnimplementedDeckardServer struct {
}

UnimplementedDeckardServer must be embedded to have forward compatible implementations.

func (UnimplementedDeckardServer) Ack 

func (UnimplementedDeckardServer) Ack(context.Context, *AckRequest) (*AckResponse, error)

func (UnimplementedDeckardServer) Add 

func (UnimplementedDeckardServer) Add(context.Context, *AddRequest) (*AddResponse, error)

func (UnimplementedDeckardServer) Count 

func (UnimplementedDeckardServer) Count(context.Context, *CountRequest) (*CountResponse, error)

func (UnimplementedDeckardServer) EditQueue 

func (UnimplementedDeckardServer) EditQueue(context.Context, *EditQueueRequest) (*EditQueueResponse, error)

func (UnimplementedDeckardServer) Flush 

func (UnimplementedDeckardServer) Flush(context.Context, *FlushRequest) (*FlushResponse, error)

func (UnimplementedDeckardServer) GetById 

func (UnimplementedDeckardServer) GetById(context.Context, *GetByIdRequest) (*GetByIdResponse, error)

func (UnimplementedDeckardServer) GetQueue 

func (UnimplementedDeckardServer) GetQueue(context.Context, *GetQueueRequest) (*GetQueueResponse, error)

func (UnimplementedDeckardServer) Nack 

func (UnimplementedDeckardServer) Nack(context.Context, *AckRequest) (*AckResponse, error)

func (UnimplementedDeckardServer) Pull 

func (UnimplementedDeckardServer) Pull(context.Context, *PullRequest) (*PullResponse, error)

func (UnimplementedDeckardServer) Remove 

func (UnimplementedDeckardServer) Remove(context.Context, *RemoveRequest) (*RemoveResponse, error)

type UnsafeDeckardServer 

type UnsafeDeckardServer interface {
	// contains filtered or unexported methods
}

UnsafeDeckardServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to DeckardServer will result in compilation errors.

Source Files

View all Source files

Directories

Path Synopsis
examples
golang
internal
audit
cmd/deckard
config
logger
messagepool
messagepool/cache
messagepool/entities
messagepool/queue
messagepool/storage
messagepool/utils
metrics
project
service
shutdown
trace

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.