firehose

README

hedwig-firehose

Firehose can be used to archive all your messages for backup / replay / debugging purpose

Documentation

Index

• Constants
• Variables
• type Clock
  • func (this *Clock) Change(time time.Time)
  • func (this *Clock) Now() time.Time
• type Firehose
  • func NewFirehose(consumerBackend hedwig.ConsumerBackend, encoderDecoder hedwig.EncoderDecoder, ...) (*Firehose, error)
  • func (fp *Firehose) IsLeader(ctx context.Context) (bool, error)
  • func (fp *Firehose) RunFirehose(ctx context.Context) error
  • func (fp *Firehose) RunFollower(ctx context.Context) error
  • func (fp *Firehose) RunLeader(ctx context.Context) error
• type LeaderFileExists
  • func (e LeaderFileExists) Error() string
• type MsgToFilePrefix
• type ProcessSettings
• type ReceivedMessage
• type StorageBackend
• type UserCreatedV1
  • func (*UserCreatedV1) Descriptor() ([]byte, []int)
  • func (x *UserCreatedV1) GetUserId() string
  • func (*UserCreatedV1) ProtoMessage()
  • func (x *UserCreatedV1) ProtoReflect() protoreflect.Message
  • func (x *UserCreatedV1) Reset()
  • func (x *UserCreatedV1) String() string

Constants

const DefaultVisibilityTimeoutS = time.Second * 20

Variables

var File_schema_proto protoreflect.FileDescriptor

Types

type Clock

type Clock struct {
	// If non-nil this is a synthetic time for testing
	Instant time.Time
	// contains filtered or unexported fields
}

func (*Clock) Change

func (this *Clock) Change(time time.Time)

func (*Clock) Now

func (this *Clock) Now() time.Time

type Firehose

type Firehose struct {
	StorageBackend StorageBackend

	HedwigFirehose *hedwig.Firehose

	Clock *Clock
	// contains filtered or unexported fields
}

func NewFirehose

func NewFirehose(consumerBackend hedwig.ConsumerBackend, encoderDecoder hedwig.EncoderDecoder, msgList []hedwig.MessageTypeMajorVersion, filePrefixes []string, mtfp MsgToFilePrefix, storageBackend StorageBackend, listenRequest hedwig.ListenRequest, consumerSettings gcp.Settings, processSettings ProcessSettings, logger hedwig.Logger) (*Firehose, error)

func (*Firehose) IsLeader

func (fp *Firehose) IsLeader(ctx context.Context) (bool, error)

func (*Firehose) RunFirehose

func (fp *Firehose) RunFirehose(ctx context.Context) error

RunFirehose starts a Firehose running in leader or follower mode

func (*Firehose) RunFollower

func (fp *Firehose) RunFollower(ctx context.Context) error

func (*Firehose) RunLeader

func (fp *Firehose) RunLeader(ctx context.Context) error

type LeaderFileExists

type LeaderFileExists struct{}

func (LeaderFileExists) Error

func (e LeaderFileExists) Error() string

type MsgToFilePrefix

type MsgToFilePrefix func(message *hedwig.Message) (string, error)

MsgToFilePrefix outputs the fileprefix (should be one of fp.fileprefixes) in StagingBucket and OutputBucket for a given hedwig message

type ProcessSettings

type ProcessSettings struct {
	// interval when leader moves files to final bucket
	ScrapeInterval int

	// interval when follower flushes to staging bucket
	FlushAfter int

	// bucket where leader file is saved
	MetadataBucket string

	// bucket where follower put intermediate files to be moved by leader
	StagingBucket string

	// final bucket for firehose files
	OutputBucket string

	// timeout before determining if node is a leader panics
	AcquireRoleTimeout int
}

type ReceivedMessage

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

type StorageBackend

type StorageBackend interface {
	// CreateWriter returns a writer for specified uploadlocation
	CreateWriter(ctx context.Context, uploadBucket string, uploadLocation string) (io.WriteCloser, error)

	// CreateReader returns a reader for specified uploadlocation
	CreateReader(ctx context.Context, uploadBucket string, uploadLocation string) (io.ReadCloser, error)

	// ListFilesPrefix should list all objects with a certain prefix
	ListFilesPrefix(ctx context.Context, bucket string, prefix string) ([]string, error)

	// DeleteFile deletes the object at the specified location
	DeleteFile(ctx context.Context, bucket string, location string) error

	// GetNodeId returns the id of the node/machine running the firehose process
	GetNodeId(ctx context.Context) string

	// GetDeploymentId returns the id of the deployment version of firehose currently running
	GetDeploymentId(ctx context.Context) string

	// WriteLeaderFile should return LeaderFileExists error if the leader file already exists fileContents should be json string of leaderFile
	WriteLeaderFile(ctx context.Context, metadataBucket string, fileContents []byte) error
}

StorageBackend is used for interacting with storage

type UserCreatedV1

type UserCreatedV1 struct {
	UserId *string `protobuf:"bytes,1,req,name=user_id,json=userId" json:"user_id,omitempty"`
	// contains filtered or unexported fields
}

func (*UserCreatedV1) Descriptor

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

Deprecated: Use UserCreatedV1.ProtoReflect.Descriptor instead.

func (*UserCreatedV1) GetUserId

func (x *UserCreatedV1) GetUserId() string

func (*UserCreatedV1) ProtoMessage

func (*UserCreatedV1) ProtoMessage()

func (*UserCreatedV1) ProtoReflect

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

func (*UserCreatedV1) Reset

func (x *UserCreatedV1) Reset()

func (*UserCreatedV1) String

func (x *UserCreatedV1) String() string