agents

Documentation

Index

• Constants
• Variables
• func CheckKafkaConn()
• func CreateConnection()
• func CreateJsonHeader(logMsg []byte, containerName string) []byte
• func Exists(name string) bool
• func GetConnection() (net.Conn, error)
• func GetContainerName(filename string) (string, error)
• func GetFileSize(fileName string) int64
• func GetFiles() []os.FileInfo
• func GetLoggingEndpoint() (string, error)
• func GetLoggingListenPort(port string) string
• func GetLoggingSvc(svc string) string
• func HandleCIMTermination()
• func HandleNatsSubscriptions()
• func InitialiseTCPLoggingParams()
• func LogModeKafka(logMsg []byte, fileName string, s *NatsSubscriber)
• func LogModeSTDOUT(logmsg []byte, containerName string)
• func LogModeTCP(logMsg []byte, fileName string, s *NatsSubscriber)
• func NewLogger(fileName string, conf types.Lmaas) *lumberjack.Logger
• func Push(ctx context.Context, key, value []byte, k *KafkaPublisher, ...) (err error)
• func Readln(r *bufio.Reader) ([]byte, error)
• func RecreateLumberjackObj()
• func Retransmit(files []os.FileInfo)
• func UpdateConfigCommitDetails(commitDetails map[string]string, cId int, cmType ...string) (err error)
• func ValidatePort(port string) bool
• func ValidateSvcUrl(url string) bool
• func WatchConfigChange(etcd *clientv3.Client, etcdWatchKey, appPort string)
• func WatchDayOneConfig()
• func WatchFilesRotation()
• func WatchLogDirectory()
• type JsonLog
• type KafkaPublisher
  • func IntialiseKafkaProducer() (*KafkaPublisher, error)
  • func NewKafkaPublisher(subject string, w *kafka.Writer) *KafkaPublisher
  • func (k KafkaPublisher) HandleMessages(msg *nats.Msg)
  • func (k KafkaPublisher) Name() string
• type Kubernetes
• type NatsSubscriber
  • func NewNatsSubscription(subject string) *NatsSubscriber
  • func (s *NatsSubscriber) FlushBufferedLogsToFile()
  • func (s *NatsSubscriber) FlushDataBeforeExit()
  • func (s *NatsSubscriber) HandleMessages(msg *nats.Msg)
  • func (s *NatsSubscriber) Name() string
  • func (s *NatsSubscriber) StartMonitoringFlushLog(flushedfileName string)
• type SubscribeHandler
• type TickerSyncMap
  • func (m *TickerSyncMap) Load(key string) *fileTicker
  • func (m *TickerSyncMap) Store(key string, value *fileTicker)

Constants

const DIR = "/opt/logs/"

Variables

var (
	KafkaPublisherEvent *KafkaPublisher

	AppConfigured             bool
	InitialNotConfiguredEvent bool
)

var (
	KafkaLogConn  *KafkaPublisher
	LogFormatJson bool
)

var (
	TCPcon       *net.TCPConn
	LogConnExist bool

	EnableDaemonsetPush bool

	EnableTCPLogging bool

	Address string
)

var (
	DayOneConfig string
)

Functions

func CheckKafkaConn

func CheckKafkaConn()

Create kafka connection when it goes down

func CreateConnection

func CreateConnection()

func CreateJsonHeader

func CreateJsonHeader(logMsg []byte, containerName string) []byte

func Exists

func Exists(name string) bool

Exists reports whether the named file or directory exists.

func GetConnection

func GetConnection() (net.Conn, error)

func GetContainerName

func GetContainerName(filename string) (string, error)

func GetFileSize

func GetFileSize(fileName string) int64

func GetFiles

func GetFiles() []os.FileInfo

func GetLoggingEndpoint

func GetLoggingEndpoint() (string, error)

func GetLoggingListenPort

func GetLoggingListenPort(port string) string

func GetLoggingSvc

func GetLoggingSvc(svc string) string

func HandleCIMTermination

func HandleCIMTermination()

HandleCIMTermination graceful termination

func HandleNatsSubscriptions

func HandleNatsSubscriptions()

HandleNatsSubscriptions handles the topic subscribed to nats

func InitialiseTCPLoggingParams

func InitialiseTCPLoggingParams()

func LogModeKafka

func LogModeKafka(logMsg []byte, fileName string, s *NatsSubscriber)

func LogModeSTDOUT

func LogModeSTDOUT(logmsg []byte, containerName string)

func LogModeTCP

func LogModeTCP(logMsg []byte, fileName string, s *NatsSubscriber)

func NewLogger

func NewLogger(fileName string, conf types.Lmaas) *lumberjack.Logger

NewLogger create a new logger obeject

func Push

func Push(ctx context.Context, key, value []byte, k *KafkaPublisher, header []kafka.Header) (err error)

Push messages to kafka

func Readln

func Readln(r *bufio.Reader) ([]byte, error)

func RecreateLumberjackObj

func RecreateLumberjackObj()

func Retransmit

func Retransmit(files []os.FileInfo)

func UpdateConfigCommitDetails

func UpdateConfigCommitDetails(commitDetails map[string]string, cId int, cmType ...string) (err error)

func ValidatePort

func ValidatePort(port string) bool

func ValidateSvcUrl

func ValidateSvcUrl(url string) bool

func WatchConfigChange

func WatchConfigChange(etcd *clientv3.Client, etcdWatchKey, appPort string)

WatchConfigChange for configuration

func WatchDayOneConfig

func WatchDayOneConfig()

WatchDayOneConfig for configuration

func WatchFilesRotation

func WatchFilesRotation()

func WatchLogDirectory

func WatchLogDirectory()

Watch Log directory if it contains logs

Types

type JsonLog

type JsonLog struct {
	Containername string     `json:"container_name"`
	Log           string     `json:"log"`
	K8s           Kubernetes `json:"kubernetes"`
}

type KafkaPublisher

type KafkaPublisher struct {
	Context   context.Context
	Subject   string
	Namespace string
	PodID     string
	Writer    *kafka.Writer
}

KafkaPublisher struct

func IntialiseKafkaProducer

func IntialiseKafkaProducer() (*KafkaPublisher, error)

Initialise kafka producer

func NewKafkaPublisher

func NewKafkaPublisher(subject string, w *kafka.Writer) *KafkaPublisher

NewKafkaPublisher initialize

func (KafkaPublisher) HandleMessages

func (k KafkaPublisher) HandleMessages(msg *nats.Msg)

HandleMessages handles messages in the NATS

func (KafkaPublisher) Name

func (k KafkaPublisher) Name() string

Name returns the name of handler

type Kubernetes

type Kubernetes struct {
	Namespace string `json:"namespace_name"`
	Podname   string `json:"pod_name"`
}

type NatsSubscriber

type NatsSubscriber struct {
	Subject    string
	QueueGroup string

	PodID        string
	Namespace    string
	FileConfig   types.Lmaas
	FlushTimeout time.Duration
	// contains filtered or unexported fields
}

func NewNatsSubscription

func NewNatsSubscription(subject string) *NatsSubscriber

NewNatsSubscription return new nats subscription

func (*NatsSubscriber) FlushBufferedLogsToFile

func (s *NatsSubscriber) FlushBufferedLogsToFile()

func (*NatsSubscriber) FlushDataBeforeExit

func (s *NatsSubscriber) FlushDataBeforeExit()

FlushDataBeforeExit flush data

func (*NatsSubscriber) HandleMessages

func (s *NatsSubscriber) HandleMessages(msg *nats.Msg)

HandleMessages handles the log messages

func (*NatsSubscriber) Name

func (s *NatsSubscriber) Name() string

Name of the handler

func (*NatsSubscriber) StartMonitoringFlushLog

func (s *NatsSubscriber) StartMonitoringFlushLog(flushedfileName string)

StartMonitoringFlushLog start monitorting the flush data

type SubscribeHandler

type SubscribeHandler interface {
	HandleMessages(msg *nats.Msg)
	Name() string
}

SubscribeHandler struct

type TickerSyncMap

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

func (*TickerSyncMap) Load

func (m *TickerSyncMap) Load(key string) *fileTicker

func (*TickerSyncMap) Store

func (m *TickerSyncMap) Store(key string, value *fileTicker)