logs

Documentation

Index

• Constants
• Variables
• type Consumer
• type ConsumerStats
• type LogsServer
  • func NewLogsServer(repo repository.Factory, state state.Interface) *LogsServer
  • func (s LogsServer) Logs(req *pb.LogRequest, stream pb.LogsService_LogsServer) error
• type LogsService
  • func NewLogsService(nats *nats.Conn, js jetstream.JetStream, state state.Interface, ...) *LogsService
  • func (ls *LogsService) AddAdapter(a adapter.Adapter)
  • func (ls *LogsService) GetConsumersStats(ctx context.Context) (stats ConsumerStats)
  • func (ls *LogsService) Run(ctx context.Context) (err error)
  • func (ls *LogsService) RunGRPCServer(ctx context.Context) error
  • func (ls *LogsService) RunHealthCheckHandler(ctx context.Context) error
  • func (ls *LogsService) Shutdown(ctx context.Context) (err error)
  • func (ls *LogsService) WithGrpcAddress(address string) *LogsService
  • func (ls *LogsService) WithHttpAddress(address string) *LogsService
  • func (ls *LogsService) WithLogsRepositoryFactory(f repository.Factory) *LogsService
  • func (ls *LogsService) WithPauseInterval(duration time.Duration) *LogsService
  • func (ls *LogsService) WithRandomPort() *LogsService

Constants

const (
	StatusPending  byte = 1
	StatusFinished byte = 2

	StreamPrefix = "log"

	StartQueue = "logsstart"
	StopQueue  = "logsstop"

	LogStartSubject = "events.logs.start"
	LogStopSubject  = "events.logs.stop"
)

const (
	DefaultHttpAddress = ":8080"
	DefaultGrpcAddress = ":9090"
)

Variables

var (
	StartSubjects = map[string]string{
		"test":    testkube.TestStartSubject,
		"generic": LogStartSubject,
	}

	StopSubjects = map[string]string{
		"test":    testkube.TestStopSubject,
		"generic": LogStopSubject,
	}
)

Types

type Consumer

type Consumer struct {
	// Name of the consumer
	Name string
	// Context is a consumer context you can call Stop() method on it when no more messages are expected
	Context jetstream.ConsumeContext
	// Instance is a NATS consumer instance
	Instance jetstream.Consumer
}

type ConsumerStats

type ConsumerStats struct {
	Count int
	Names []string
}

type LogsServer

type LogsServer struct {
	pb.UnimplementedLogsServiceServer
	// contains filtered or unexported fields
}

func NewLogsServer

func NewLogsServer(repo repository.Factory, state state.Interface) *LogsServer

func (LogsServer) Logs

func (s LogsServer) Logs(req *pb.LogRequest, stream pb.LogsService_LogsServer) error

type LogsService

type LogsService struct {
	Ready chan struct{}
	// contains filtered or unexported fields
}

func NewLogsService

func NewLogsService(nats *nats.Conn, js jetstream.JetStream, state state.Interface, stream client.Stream) *LogsService

func (*LogsService) AddAdapter

func (ls *LogsService) AddAdapter(a adapter.Adapter)

AddAdapter adds new adapter to logs service adapters will be configred based on given mode e.g. cloud mode will get cloud adapter to store logs directly on the cloud

func (*LogsService) GetConsumersStats

func (ls *LogsService) GetConsumersStats(ctx context.Context) (stats ConsumerStats)

func (*LogsService) Run

func (ls *LogsService) Run(ctx context.Context) (err error)

func (*LogsService) RunGRPCServer

func (ls *LogsService) RunGRPCServer(ctx context.Context) error

TODO handle TLS

func (*LogsService) RunHealthCheckHandler

func (ls *LogsService) RunHealthCheckHandler(ctx context.Context) error

RunHealthCheckHandler is a handler for health check events we need HTTP as GRPC probes starts from Kubernetes 1.25

func (*LogsService) Shutdown

func (ls *LogsService) Shutdown(ctx context.Context) (err error)

func (*LogsService) WithGrpcAddress

func (ls *LogsService) WithGrpcAddress(address string) *LogsService

func (*LogsService) WithHttpAddress

func (ls *LogsService) WithHttpAddress(address string) *LogsService

func (*LogsService) WithLogsRepositoryFactory

func (ls *LogsService) WithLogsRepositoryFactory(f repository.Factory) *LogsService

func (*LogsService) WithPauseInterval

func (ls *LogsService) WithPauseInterval(duration time.Duration) *LogsService

func (*LogsService) WithRandomPort

func (ls *LogsService) WithRandomPort() *LogsService