agents

package

v0.0.0-...-f374617 Latest Latest Go to latest Published: Apr 19, 2021 License: Apache-2.0 Imports: 35 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/xgvela/cim

Links

Open Source Insights

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 ¶

View Source

const DIR = "/opt/logs/"

Variables ¶

View Source

var (
	KafkaPublisherEvent *KafkaPublisher

	AppConfigured             bool
	InitialNotConfiguredEvent bool
)

View Source

var (
	KafkaLogConn  *KafkaPublisher
	LogFormatJson bool
)

View Source

var (
	TCPcon       *net.TCPConn
	LogConnExist bool

	EnableDaemonsetPush bool

	EnableTCPLogging bool

	Address string
)

View Source

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)

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL