cloudwatchlogspublisher

Documentation

Overview

Package cloudwatchlogspublisher is responsible for pulling logs from the log queue and publishing them to cloudwatch

Index

• Constants
• type CloudWatchLogsService
  • func NewCloudWatchLogsService(context context.T) *CloudWatchLogsService
  • func NewCloudWatchLogsServiceWithCredentials(context context.T, id, secret string) *CloudWatchLogsService
  • func (service *CloudWatchLogsService) CreateLogGroup(logGroup string) (err error)
  • func (service *CloudWatchLogsService) CreateLogStream(logGroup, logStream string) (err error)
  • func (service *CloudWatchLogsService) CreateNewServiceIfUnHealthy()
  • func (service *CloudWatchLogsService) DescribeLogGroups(logGroupPrefix, nextToken string) (response *cloudwatchlogs.DescribeLogGroupsOutput, err error)
  • func (service *CloudWatchLogsService) DescribeLogStreams(logGroup, logStreamPrefix, nextToken string) (response *cloudwatchlogs.DescribeLogStreamsOutput, err error)
  • func (service *CloudWatchLogsService) GetIsUploadComplete() bool
  • func (service *CloudWatchLogsService) GetSequenceTokenForStream(logGroupName, logStreamName string) (sequenceToken *string)
  • func (service *CloudWatchLogsService) IsLogGroupEncryptedWithKMS(logGroup *cloudwatchlogs.LogGroup) (bool, error)
  • func (service *CloudWatchLogsService) IsLogGroupPresent(logGroup string) (bool, *cloudwatchlogs.LogGroup)
  • func (service *CloudWatchLogsService) IsLogStreamPresent(logGroupName, logStreamName string) bool
  • func (service *CloudWatchLogsService) PutLogEvents(messages []*cloudwatchlogs.InputLogEvent, logGroup, logStream string, ...) (nextSequenceToken *string, err error)
  • func (service *CloudWatchLogsService) SetCloudWatchMessage(eventVersion string, awsRegion string, targetId string, runAsUser string, ...)
  • func (service *CloudWatchLogsService) SetIsFileComplete(val bool)
  • func (service *CloudWatchLogsService) StreamData(logGroupName string, logStreamName string, absoluteFilePath string, ...) (success bool)
• type CloudWatchMessage
• type CloudWatchPublisher
  • func NewCloudWatchPublisher(context context.T) *CloudWatchPublisher
  • func (cloudwatchPublisher *CloudWatchPublisher) CloudWatchLogsEventsListener()
  • func (cloudwatchPublisher *CloudWatchPublisher) Init()
  • func (cloudwatchPublisher *CloudWatchPublisher) Start()
  • func (cloudwatchPublisher *CloudWatchPublisher) Stop()
• type ICloudWatchPublisher
• type Target
• type UserIdentity

Constants

const (
	UploadFrequency  = 1 * time.Second
	NewLineCharacter = '\n'

	// Event size - https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/cloudwatch_limits_cwl.html
	MessageLengthThresholdInBytes = 200 * 1000
	// json.Marshal can inflate streamed event message by up to ~6 times the size of the message, depending on message contents.
	// Threshold is reduced here to 1/6 size to account for this.
	// https://go.dev/play/p/G3RalE_BUEL
	StreamMessageLengthThresholdInBytes = 33000
)

Types

type CloudWatchLogsService

type CloudWatchLogsService struct {
	CloudWatchMessage CloudWatchMessage
	// contains filtered or unexported fields
}

CloudWatchLogsService encapsulates the client and stop policy as a wrapper to call the cloudwatchlogs API

func NewCloudWatchLogsService

func NewCloudWatchLogsService(context context.T) *CloudWatchLogsService

NewCloudWatchLogsService Creates a new instance of the CloudWatchLogsService

func NewCloudWatchLogsServiceWithCredentials

func NewCloudWatchLogsServiceWithCredentials(context context.T, id, secret string) *CloudWatchLogsService

NewCloudWatchLogsServiceWithCredentials Creates a new instance of the CloudWatchLogsService using credentials from the Id and Secret passed

func (*CloudWatchLogsService) CreateLogGroup

func (service *CloudWatchLogsService) CreateLogGroup(logGroup string) (err error)

CreateLogGroup calls the CreateLogGroup API to create a log group

func (*CloudWatchLogsService) CreateLogStream

func (service *CloudWatchLogsService) CreateLogStream(logGroup, logStream string) (err error)

CreateLogStream calls the CreateLogStream API to create log stream within the specified log group

func (*CloudWatchLogsService) CreateNewServiceIfUnHealthy

func (service *CloudWatchLogsService) CreateNewServiceIfUnHealthy()

CreateNewServiceIfUnHealthy checks service healthy and create new service if original is unhealthy

func (*CloudWatchLogsService) DescribeLogGroups

func (service *CloudWatchLogsService) DescribeLogGroups(logGroupPrefix, nextToken string) (response *cloudwatchlogs.DescribeLogGroupsOutput, err error)

DescribeLogGroups calls the DescribeLogGroups API to get the details of log groups of account

func (*CloudWatchLogsService) DescribeLogStreams

func (service *CloudWatchLogsService) DescribeLogStreams(logGroup, logStreamPrefix, nextToken string) (response *cloudwatchlogs.DescribeLogStreamsOutput, err error)

DescribeLogStreams calls the DescribeLogStreams API to get the details of the log streams present

func (*CloudWatchLogsService) GetIsUploadComplete

func (service *CloudWatchLogsService) GetIsUploadComplete() bool

func (*CloudWatchLogsService) GetSequenceTokenForStream

func (service *CloudWatchLogsService) GetSequenceTokenForStream(logGroupName, logStreamName string) (sequenceToken *string)

GetSequenceTokenForStream returns the current sequence token for the stream specified

func (*CloudWatchLogsService) IsLogGroupEncryptedWithKMS

func (service *CloudWatchLogsService) IsLogGroupEncryptedWithKMS(logGroup *cloudwatchlogs.LogGroup) (bool, error)

IsLogGroupEncryptedWithKMS return true if the log group is encrypted with KMS key.

func (*CloudWatchLogsService) IsLogGroupPresent

func (service *CloudWatchLogsService) IsLogGroupPresent(logGroup string) (bool, *cloudwatchlogs.LogGroup)

IsLogGroupPresent checks and returns true when the log group is present

func (*CloudWatchLogsService) IsLogStreamPresent

func (service *CloudWatchLogsService) IsLogStreamPresent(logGroupName, logStreamName string) bool

IsLogStreamPresent checks and returns true when the log stream is present

func (*CloudWatchLogsService) PutLogEvents

func (service *CloudWatchLogsService) PutLogEvents(messages []*cloudwatchlogs.InputLogEvent, logGroup, logStream string, sequenceToken *string) (nextSequenceToken *string, err error)

PutLogEvents calls the PutLogEvents API to push messages to CloudWatchLogs

func (*CloudWatchLogsService) SetCloudWatchMessage

func (service *CloudWatchLogsService) SetCloudWatchMessage(
	eventVersion string,
	awsRegion string,
	targetId string,
	runAsUser string,
	sessionId string,
	sessionOwner string)

SetCloudWatchMessage initializes CloudWatchMessage

func (*CloudWatchLogsService) SetIsFileComplete

func (service *CloudWatchLogsService) SetIsFileComplete(val bool)

func (*CloudWatchLogsService) StreamData

func (service *CloudWatchLogsService) StreamData(
	logGroupName string,
	logStreamName string,
	absoluteFilePath string,
	isFileComplete bool,
	isLogStreamCreated bool,
	fileCompleteSignal chan bool,
	cleanupControlCharacters bool,
	structuredLogs bool) (success bool)

StreamData streams data from the absoluteFilePath file to cloudwatch logs.

type CloudWatchMessage

type CloudWatchMessage struct {
	EventVersion *string       `json:"eventVersion"`
	EventTime    *string       `json:"eventTime"`
	AwsRegion    *string       `json:"awsRegion"`
	Target       *Target       `json:"target"`
	UserIdentity *UserIdentity `json:"userIdentity"`
	RunAsUser    *string       `json:"runAsUser"`
	SessionId    *string       `json:"sessionId"`
	SessionData  []*string     `json:"sessionData"`
	// contains filtered or unexported fields
}

CloudWatchMessage captures all the information that are published in an event for streaming logs

type CloudWatchPublisher

type CloudWatchPublisher struct {
	QueuePollingInterval time.Duration // The interval after which the publisher polls the queue
	QueuePollingWaitTime time.Duration // The duration for which the publisher blocks while polling. For negative value will wait until enqueue
	// contains filtered or unexported fields
}

CloudWatchPublisher wrapper to publish logs to cloudwatchlogs

func NewCloudWatchPublisher

func NewCloudWatchPublisher(context context.T) *CloudWatchPublisher

func (*CloudWatchPublisher) CloudWatchLogsEventsListener

func (cloudwatchPublisher *CloudWatchPublisher) CloudWatchLogsEventsListener()

CloudWatchLogsEventsListener listens to cloudwatchlogs events channel

func (*CloudWatchPublisher) Init

func (cloudwatchPublisher *CloudWatchPublisher) Init()

Init initializes the publisher

func (*CloudWatchPublisher) Start

func (cloudwatchPublisher *CloudWatchPublisher) Start()

Start starts the publisher to consume messages from the queue

func (*CloudWatchPublisher) Stop

func (cloudwatchPublisher *CloudWatchPublisher) Stop()

Stop called to stop the publisher

type ICloudWatchPublisher

type ICloudWatchPublisher interface {
	Init() (err error)
	Start()
	Stop()
}

ICloudWatchPublisher interface for publishing logs to cloudwatchlogs

type Target

type Target struct {
	Id *string `json:"id"`
	// contains filtered or unexported fields
}

Target represents id of the target

type UserIdentity

type UserIdentity struct {
	Arn *string `json:"arn"`
	// contains filtered or unexported fields
}

UserIdentity represents iam arn of the requester