minio

README

Kubemq Minio/S3 Source Connector

Kubemq Minio/S3 source connector allows services using kubemq server to sync minio objects to remote location.

Prerequisites

The following are required to run the minio source connector:

• kubemq cluster
• minio cluster / AWS s3 service
• kubemq-sources deployment

Configuration

Minio source connector configuration properties:

Properties Key	Required	Description	Example
endpoint	yes	minio host address	"localhost:9000"
use_ssl	no	set connection ssl	"true"
access_key_id	yes	set access key id	"minio"
secret_access_key	yes	set secret access key	"minio123"
folders	yes	set list of folders to watch	"/"
target_type	yes	set remote target sync type	"aws", "gcp", "minio", "filesystem","hdfs","azure","pass-through"
bucket_name	yes	set source bucket	"bucket"
concurrency	no	set sending concurrency	"1"
scan_interval	no	set bucket scan interval in sec	"5"

Example:

bindings:
- name: minio
  source:
    kind: storage.minio
    properties:
      endpoint: "localhost:9000"
      use_ssl: "false"
      access_key_id: "minio"
      secret_access_key: "minio123"
      folders: 'folder1,folder2/sub1/sub2'
      target_type: filesystem
      bucket_name: bucket
      concurrency: 1
      scan_interval: 5
  target:
    kind: kubemq.queue
    properties:
      address: localhost:50000
      channel: queue.minio
  properties: {}

Documentation

Index

• func Connector() *common.Connector
• type Client
  • func New() *Client
  • func (c *Client) Connector() *common.Connector
  • func (c *Client) Init(ctx context.Context, cfg config.Spec, log *logger.Logger) error
  • func (c *Client) Start(ctx context.Context, target middleware.Middleware) error
  • func (c *Client) Stop() error
• type SourceFile
  • func NewSourceFile(c *minio.Client, bucket string, obj minio.ObjectInfo) *SourceFile
  • func (s *SourceFile) Delete(ctx context.Context) error
  • func (s *SourceFile) Do(ctx context.Context) error
  • func (s *SourceFile) FileDir() string
  • func (s *SourceFile) FileName() string
  • func (s *SourceFile) FullPath() string
  • func (s *SourceFile) Hash() string
  • func (s *SourceFile) Load(ctx context.Context) ([]byte, error)
  • func (s *SourceFile) Metadata() string
  • func (s *SourceFile) Request(ctx context.Context, bucketType string, bucketName string) (*types.Request, error)
  • func (s *SourceFile) RootDir() string
• type TargetsMetadata
  • func NewTargetMetadata() TargetsMetadata
  • func (m TargetsMetadata) Set(key, value string) TargetsMetadata
  • func (m TargetsMetadata) String() string
• type TargetsRequest
  • func NewTargetsRequest() *TargetsRequest
  • func ParseRequest(body []byte) (*TargetsRequest, error)
  • func (r *TargetsRequest) MarshalBinary() []byte
  • func (r *TargetsRequest) SetData(value []byte) *TargetsRequest
  • func (r *TargetsRequest) SetMetadata(value TargetsMetadata) *TargetsRequest
  • func (r *TargetsRequest) SetMetadataKeyValue(key, value string) *TargetsRequest
  • func (r *TargetsRequest) Size() float64

Functions

func Connector

func Connector() *common.Connector

Types

type Client

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

func New

func New() *Client

func (*Client) Connector

func (c *Client) Connector() *common.Connector

func (*Client) Init

func (c *Client) Init(ctx context.Context, cfg config.Spec, log *logger.Logger) error

func (*Client) Start

func (c *Client) Start(ctx context.Context, target middleware.Middleware) error

func (*Client) Stop

func (c *Client) Stop() error

type SourceFile

type SourceFile struct {
	Object minio.ObjectInfo
	Bucket string
	// contains filtered or unexported fields
}

func NewSourceFile

func NewSourceFile(c *minio.Client, bucket string, obj minio.ObjectInfo) *SourceFile

func (*SourceFile) Delete

func (s *SourceFile) Delete(ctx context.Context) error

func (*SourceFile) Do

func (s *SourceFile) Do(ctx context.Context) error

func (*SourceFile) FileDir

func (s *SourceFile) FileDir() string

func (*SourceFile) FileName

func (s *SourceFile) FileName() string

func (*SourceFile) FullPath

func (s *SourceFile) FullPath() string

func (*SourceFile) Hash

func (s *SourceFile) Hash() string

func (*SourceFile) Load

func (s *SourceFile) Load(ctx context.Context) ([]byte, error)

func (*SourceFile) Metadata

func (s *SourceFile) Metadata() string

func (*SourceFile) Request

func (s *SourceFile) Request(ctx context.Context, bucketType string, bucketName string) (*types.Request, error)

func (*SourceFile) RootDir

func (s *SourceFile) RootDir() string

type TargetsMetadata

type TargetsMetadata map[string]string

func NewTargetMetadata

func NewTargetMetadata() TargetsMetadata

func (TargetsMetadata) Set

func (m TargetsMetadata) Set(key, value string) TargetsMetadata

func (TargetsMetadata) String

func (m TargetsMetadata) String() string

type TargetsRequest

type TargetsRequest struct {
	Metadata TargetsMetadata `json:"metadata,omitempty"`
	Data     []byte          `json:"data,omitempty"`
}

func NewTargetsRequest

func NewTargetsRequest() *TargetsRequest

func ParseRequest

func ParseRequest(body []byte) (*TargetsRequest, error)

func (*TargetsRequest) MarshalBinary

func (r *TargetsRequest) MarshalBinary() []byte

func (*TargetsRequest) SetData

func (r *TargetsRequest) SetData(value []byte) *TargetsRequest

func (*TargetsRequest) SetMetadata

func (r *TargetsRequest) SetMetadata(value TargetsMetadata) *TargetsRequest

func (*TargetsRequest) SetMetadataKeyValue

func (r *TargetsRequest) SetMetadataKeyValue(key, value string) *TargetsRequest

func (*TargetsRequest) Size

func (r *TargetsRequest) Size() float64