Feature heartbeat

Makefile

@@ -82,13 +82,13 @@ benchmark: ## Run benchmark
 
 ##@ Build
 
-build: ## go build
-	CGO_ENABLED=1 GOOS=${GOOS} GOARCH=${GOARCH} go build -mod=vendor -a ${extra_flags} -o loggie cmd/loggie/main.go
+build: ## go build, EXT_BUILD_TAGS=include_core would only build core package
+	CGO_ENABLED=1 GOOS=${GOOS} GOARCH=${GOARCH} go build -tags ${EXT_BUILD_TAGS} -mod=vendor -a ${extra_flags} -o loggie cmd/loggie/main.go
 
 ##@ Build(without sqlite)
 
-build-in-badger: ## go build without sqlite
-	GOOS=${GOOS} GOARCH=${GOARCH} go build -tags driver_badger -mod=vendor -a -ldflags '-X github.com/loggie-io/loggie/pkg/core/global._VERSION_=${TAG} -X github.com/loggie-io/loggie/pkg/util/persistence._DRIVER_=badger -s -w' -o loggie cmd/loggie/main.go
+build-in-badger: ## go build without sqlite, EXT_BUILD_TAGS=include_core would only build core package
+	GOOS=${GOOS} GOARCH=${GOARCH} go build -tags driver_badger,${EXT_BUILD_TAGS} -mod=vendor -a -ldflags '-X github.com/loggie-io/loggie/pkg/core/global._VERSION_=${TAG} -X github.com/loggie-io/loggie/pkg/util/persistence._DRIVER_=badger -s -w' -o loggie cmd/loggie/main.go
 
 ##@ Images
 
@@ -98,7 +98,7 @@ docker-build: ## Docker build -t ${REPO}:${TAG}, try: make docker-build REPO=<Yo
 docker-push: ## Docker push ${REPO}:${TAG}
 	docker push ${REPO}:${TAG}
 
-docker-multi-arch: ## Docker buildx, try: make docker-build REPO=<YourRepoHost>, ${TAG} generated by git
+docker-multi-arch: ## Docker buildx, try: make docker-multi-arch REPO=<YourRepoHost>, ${TAG} generated by git
 	docker buildx build --platform linux/amd64,linux/arm64 -t ${REPO}:${TAG} . --push
 
 LOG_DIR ?= /tmp/log ## log directory

cmd/loggie/main.go

@@ -19,6 +19,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"github.com/loggie-io/loggie/pkg/heartbeat"
 	"github.com/loggie-io/loggie/pkg/ops"
 	"github.com/pkg/errors"
 	"go.uber.org/automaxprocs/maxprocs"
@@ -49,6 +50,7 @@ var (
 	pipelineConfigPath string
 	configType         string
 	nodeName           string
+	mainServer         string
 )
 
 func init() {
@@ -143,6 +145,11 @@ func main() {
 		}()
 	}
 
+	// heartbeat
+	if syscfg.Loggie.HeartbeatConfig.Address != "" {
+		go heartbeat.Start(syscfg.Loggie.HeartbeatConfig)
+	}
+
 	log.Info("started Loggie")
 	<-stopCh
 	log.Info("shutting down Loggie")

doc/changelog/CHANGELOG-v1.md

@@ -1,3 +1,157 @@
+# Release v1.5.0
+
+### :star2: Features
+- [breaking]: The `db` in `file source` is moved to the [`loggie.yml`](https://loggie-io.github.io/docs/main/reference/global/db/). If upgrading from an earlier version to v1.5, be sure to check whether `db` has been configured for `file source`. If it is not configured, you can just ignore it, and the default value will remain compatible.
+- Added rocketmq sink (#530)
+- Added franzKafka source (#573)
+- Added kata runtime (#554)
+- `typePodFields`/`typeNodeFields` is supported in LogConfig/ClusterLogConfig  (#450)
+- sink codec support printEvents (#448)
+- Added queue in LogConfig/ClusterLogConfig  (#457)
+- Changed `olivere/elastic` to the official elasticsearch go client (#581)
+- Supported `copytruncate` in file source (#571)
+- Added `genfiles` sub command (#471)
+- Added queue in LogConfig/ClusterLogConfig queue (#457)
+- Added `sortBy` field in elasticsearch source (#473)
+- Added host VM mode with Kubernetes as the configuration center (#449) (#489)
+- New `addHostMeta` interceptor (#474)
+- Added persistence driver `badger` (#475) (#584)
+- Ignore LogConfig with sidecar injection annotation (#478)
+- Added `toStr` action in transformer interceptor (#482)
+- You can mount the root directory of a node to the Loggie container without mounting additional Loggie volumes (#460)
+- Get loggie version with api and sub command (#496) (#508)
+- Added the `worker` and the `clientId` in Kafka source  (#506) (#507)
+- Upgrade `kafka-go` version  (#506) (#567)
+- Added resultStatus in dev sink which can be used to simulate failure, drop (#531)
+- Pretty error when unmarshal yaml configuration failed (#539)
+- Added default topic if render kafka topic failed (#550)
+- Added `ignoreUnknownTopicOrPartition` in kafka sink (#560)
+- Supported multiple topics in kafka source (#548)
+- Added default index if render elasticsearch index failed (#551) (#553)
+- The default `maxOpenFds` is set to 4096 (#559)
+- Supported default `sinkRef` in kubernetes discovery (#555)
+- Added `${_k8s.clusterlogconfig}` in `typePodFields` (#569)
+- Supported omit empty fields in Kubernetes discovery (#570)
+- Optimizes `maxbytes` interceptors (#575)
+- Moved `readFromTail`, `cleanFiles`, `fdHoldTimeoutWhenInactive`, `fdHoldTimeoutWhenRemove` from watcher to outer layer in `file source` (#579) (#585)
+- Added `cheanUnfinished` in cleanFiles (#580)
+- Added `target` in `maxbyte` interceptor (#588)
+- Added `partionKey` in franzKafka (#562)
+- Added `highPrecision` in `rateLimit` interceptor (#525)
+- Supported build core components of Loggie (#598)
+- Added `addonMetaSchema` in file source (#599)
+- Added `timestamp` and `bodyKey` in source (#600)
+- Added `discoverNodesOnStart` and `discoverNodesInterval` in elasticsearch sink (#612)
+
+### :bug: Bug Fixes
+- Fixed panic when kubeEvent Series is nil (#459)
+- Upgraded `automaxprocs` version to v1.5.1 (#488)
+- Fixed set defaults failed in `fieldsUnderKey` (#513)
+- Fixed parse condition failed when contain ERROR in transformer interceptor (#514) (#515)
+- Fixed grpc batch out-of-order data streams (#517)
+- Fixed large line may cause oom (#529)
+- Fixed duplicated batchSize in queue (#533)
+- Fixed sqlite locked panic (#524)
+- Fixed command can't be used in multi-arch container (#541)
+- Fixed `logger listener` may cause block (#561) (#552)
+- Fixed `sink concurrency` deepCopy failed (#563)
+- Drop events when partial error in elasticsearch sink (#572)
+- Fix convert to string type when returning byte array type in transformer interceptor (#596)
+- Invalid pipelines will not stop all the pipelines running (#602)
+- Fixed configuration npe when query pipelines (#604)
+- Optimized elasticsearch sink request buffer (#608)
+- Add pod namespace as prefix of source name when pod is matched by ClusterLogConfig (#613)
+
+# Release v1.5.0-rc.0
+
+### :star2: Features
+- [breaking]: The `db` in `file source` is moved to the [`loggie.yml`](https://loggie-io.github.io/docs/main/reference/global/db/). If upgrading from an earlier version to v1.5, be sure to check whether `db` has been configured for `file source`. If it is not configured, you can just ignore it, and the default value will remain compatible.
+- Added rocketmq sink (#530)
+- Added franzKafka source (#573)
+- Added kata runtime (#554)
+- `typePodFields`/`typeNodeFields` is supported in LogConfig/ClusterLogConfig  (#450)
+- sink codec support printEvents (#448)
+- Added queue in LogConfig/ClusterLogConfig  (#457)
+- Changed `olivere/elastic` to the official elasticsearch go client (#581)
+- Supported `copytruncate` in file source (#571)
+- Added `genfiles` sub command (#471)
+- Added queue in LogConfig/ClusterLogConfig queue (#457)
+- Added `sortBy` field in elasticsearch source (#473)
+- Added host VM mode with Kubernetes as the configuration center (#449) (#489)
+- New `addHostMeta` interceptor (#474)
+- Added persistence driver `badger` (#475) (#584)
+- Ignore LogConfig with sidecar injection annotation (#478)
+- Added `toStr` action in transformer interceptor (#482)
+- You can mount the root directory of a node to the Loggie container without mounting additional Loggie volumes (#460)
+- Get loggie version with api and sub command (#496) (#508)
+- Added the `worker` and the `clientId` in Kafka source  (#506) (#507)
+- Upgrade `kafka-go` version  (#506) (#567)
+- Added resultStatus in dev sink which can be used to simulate failure, drop (#531)
+- Pretty error when unmarshal yaml configuration failed (#539)
+- Added default topic if render kafka topic failed (#550)
+- Added `ignoreUnknownTopicOrPartition` in kafka sink (#560)
+- Supported multiple topics in kafka source (#548)
+- Added default index if render elasticsearch index failed (#551) (#553)
+- The default `maxOpenFds` is set to 4096 (#559)
+- Supported default `sinkRef` in kubernetes discovery (#555)
+- Added `${_k8s.clusterlogconfig}` in `typePodFields` (#569)
+- Supported omit empty fields in Kubernetes discovery (#570)
+- Optimizes `maxbytes` interceptors (#575)
+- Moved `readFromTail`, `cleanFiles`, `fdHoldTimeoutWhenInactive`, `fdHoldTimeoutWhenRemove` from watcher to outer layer in `file source` (#579) (#585)
+- Added `cheanUnfinished` in cleanFiles (#580)
+- Added `target` in `maxbyte` interceptor (#588)
+- Added `partionKey` in franzKafka (#562)
+- Added `highPrecision` in `rateLimit` interceptor (#525)
+
+### :bug: Bug Fixes
+- Fixed panic when kubeEvent Series is nil (#459)
+- Upgraded `automaxprocs` version to v1.5.1 (#488)
+- Fixed set defaults failed in `fieldsUnderKey` (#513)
+- Fixed parse condition failed when contain ERROR in transformer interceptor (#514) (#515)
+- Fixed grpc batch out-of-order data streams (#517)
+- Fixed large line may cause oom (#529)
+- Fixed duplicated batchSize in queue (#533)
+- Fixed sqlite locked panic (#524)
+- Fixed command can't be used in multi-arch container (#541)
+- Fixed `logger listener` may cause block (#561) (#552)
+- Fixed `sink concurrency` deepCopy failed (#563)
+- Drop events when partial error in elasticsearch sink (#572)
+
+# Release v1.4.0
+
+### :star2: Features
+
+- Added Loggie dashboard feature for easier troubleshooting (#416)
+- Enhanced log alerting function with more flexible log alert detection rules and added alertWebhook sink (#392)
+- Added sink concurrency support for automatic adaptation based on downstream delay (#376)
+- Added franzKafka sink for users who prefer the franz kafka library (#423)
+- Added elasticsearch source (#345)
+- Added zinc sink (#254)
+- Added pulsar sink (#417)
+- Added grok action to transformer interceptor (#418)
+- Added split action to transformer interceptor (#411)
+- Added jsonEncode action to transformer interceptor (#421)
+- Added fieldsFromPath configuration to source for obtaining fields from file content (#401)
+- Added fieldsRef parameter to filesource listener for obtaining key value from fields configuration and adding to metrics as label (#402)
+- In transformer interceptor, added dropIfError support to drop event if action execution fails (#409)
+- Added info listener which currently exposes loggie_info_stat metrics and displays version label (#410)
+- Added support for customized kafka sink partition key
+- Added sasl support to Kafka source (#415)
+- Added https insecureSkipVerify support to loki sink (#422)
+- Optimized file source for large files (#430)
+- Changed default value of file source maxOpenFds to 1024 (#437)
+- ContainerRuntime can now be set to none (#439)
+- Upgraded to go 1.18 (#440)
+- Optimize the configuration parameters to remove the redundancy generated by rendering
+
+### :bug: Bug Fixes
+
+- Added source fields to filesource listener (#402)
+- Fixed issue of transformer copy action not copying non-string body (#420)
+- Added fetching of logs file from UpperDir when rootfs collection is enabled (#414)
+- Fix pipeline restart npe (#454)
+- Fix create dir soft link job (#453)
+
 # Release v1.4.0-rc.0
 
 ### :star2: Features

go.mod

@@ -61,7 +61,6 @@ require (
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/danieljoos/wincred v1.0.2 // indirect
 	github.com/dgraph-io/ristretto v0.1.1 // indirect
-	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/dvsekhvalnov/jose2go v0.0.0-20200901110807-248326c1351b // indirect
 	github.com/emirpasic/gods v1.12.0 // indirect
 	github.com/fatih/color v1.10.0 // indirect
@@ -174,8 +173,10 @@ require (
 
 require (
 	github.com/apache/rocketmq-client-go/v2 v2.1.1
+	github.com/dustin/go-humanize v1.0.0
 	github.com/elastic/go-elasticsearch/v7 v7.17.10
 	github.com/goccy/go-yaml v1.11.0
+	github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf
 	github.com/mattn/go-sqlite3 v1.11.0
 	k8s.io/metrics v0.25.4
 	sigs.k8s.io/controller-runtime v0.13.1

go.sum

@@ -548,6 +548,8 @@ github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf h1:WfD7VjIE6z8dIvMsI4/s+1qr5EL+zoIGev1BQj1eoJ8=
+github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf/go.mod h1:hyb9oH7vZsitZCiBt0ZvifOrB+qc8PS5IiilCIb87rg=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY=
 github.com/influxdata/influxdb v1.8.3/go.mod h1:JugdFhsvvI8gadxOI6noqNeeBHvWNTbfYGtiAn+2jhI=

loggie.yml

@@ -14,7 +14,7 @@ loggie:
       sink: ~
       queue: ~
       pipeline: ~
-      normalize: ~
+      sys: ~
 
   discovery:
     enabled: false
@@ -31,16 +31,23 @@ loggie:
   defaults:
     sink:
       type: dev
-    interceptors:
-      - type: schema
-        name: global
-        order: 700
-        addMeta:
-          timestamp:
-            key: "@timestamp"
     sources:
       - type: file
+        timestampKey: "@timestamp"
+        bodyKey: "message"
+        fieldsUnderRoot: true
+        addonMeta: true
+        addonMetaSchema:
+          underRoot: true
+          fields:
+            filename: "${_meta.filename}"
+            line: "${_meta.line}"
         watcher:
           maxOpenFds: 6000
   http:
     enabled: true
+
+  heartbeat:
+    address: http://127.0.0.1:80/api/heartbeat
+    interval: 30
+    nodeName: 01

pkg/control/config.go

@@ -130,11 +130,21 @@ func ReadPipelineConfigFromFile(path string, ignore FileIgnore) (*PipelineConfig
 	for _, fn := range all {
 		pipes := &PipelineConfig{}
 		unpack := cfg.UnPackFromFile(fn, pipes)
-		if err = unpack.Defaults().Validate().Do(); err != nil {
-			log.Error("invalid pipeline configs: %v, \n%s", err, unpack.Contents())
+		if err = unpack.Do(); err != nil {
+			log.Error("read pipeline configs from path %s failed: %v", path, err)
 			continue
 		}
-		pipecfgs.AddPipelines(pipes.Pipelines)
+
+		for _, p := range pipes.Pipelines {
+			pip := p
+			if err := cfg.NewUnpack(nil, &pip, nil).Defaults().Validate().Do(); err != nil {
+				// ignore invalid pipeline, but continue to read other pipelines
+				// invalid pipeline will check by reloader later
+				log.Error("pipeline: %s configs invalid: %v", p.Name, err)
+				continue
+			}
+			pipecfgs.AddPipelines([]pipeline.Config{pip})
+		}
 	}
 	return pipecfgs, nil
 }

pkg/core/reloader/http.go

@@ -17,7 +17,6 @@ limitations under the License.
 package reloader
 
 import (
-	"io/ioutil"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -49,7 +48,7 @@ func (r *reloader) readPipelineConfigHandler(writer http.ResponseWriter, request
 			continue
 		}
 
-		content, err := ioutil.ReadFile(m)
+		content, err := os.ReadFile(m)
 		if err != nil {
 			log.Warn("read config error. err: %v", err)
 			return

pkg/core/source/config.go

@@ -38,6 +38,11 @@ type Config struct {
 	FieldsFromEnv   map[string]string      `yaml:"fieldsFromEnv,omitempty"`
 	FieldsFromPath  map[string]string      `yaml:"fieldsFromPath,omitempty"`
 	Codec           *codec.Config          `yaml:"codec,omitempty"`
+
+	TimestampKey      string `yaml:"timestampKey,omitempty"`
+	TimestampLocation string `yaml:"timestampLocation,omitempty"`
+	TimestampLayout   string `yaml:"timestampLayout,omitempty"`
+	BodyKey           string `yaml:"bodyKey,omitempty"`
 }
 
 func (c *Config) DeepCopy() *Config {
@@ -82,6 +87,11 @@ func (c *Config) DeepCopy() *Config {
 		FieldsFromEnv:   newFieldsFromEnv,
 		FieldsFromPath:  newFieldsFromPath,
 		Codec:           c.Codec.DeepCopy(),
+
+		TimestampKey:      c.TimestampKey,
+		TimestampLocation: c.TimestampLocation,
+		TimestampLayout:   c.TimestampLayout,
+		BodyKey:           c.BodyKey,
 	}
 
 	return out
@@ -155,6 +165,19 @@ func (c *Config) Merge(from *Config) {
 	} else {
 		c.Codec.Merge(from.Codec)
 	}
+
+	if c.TimestampKey == "" {
+		c.TimestampKey = from.TimestampKey
+	}
+	if c.TimestampLocation == "" {
+		c.TimestampLocation = from.TimestampLocation
+	}
+	if c.TimestampLayout == "" {
+		c.TimestampLayout = from.TimestampLayout
+	}
+	if c.BodyKey == "" {
+		c.BodyKey = from.BodyKey
+	}
 }
 
 func MergeSourceList(base []*Config, from []*Config) []*Config {

pkg/core/sysconfig/config.go

@@ -25,6 +25,7 @@ import (
 	"github.com/loggie-io/loggie/pkg/core/source"
 	"github.com/loggie-io/loggie/pkg/discovery"
 	"github.com/loggie-io/loggie/pkg/eventbus"
+	"github.com/loggie-io/loggie/pkg/heartbeat"
 	"github.com/loggie-io/loggie/pkg/interceptor/maxbytes"
 	"github.com/loggie-io/loggie/pkg/interceptor/metric"
 	"github.com/loggie-io/loggie/pkg/interceptor/retry"
@@ -45,6 +46,7 @@ type Loggie struct {
 	Defaults         Defaults                    `yaml:"defaults"`
 	Db               persistence.DbConfig        `yaml:"db"`
 	ErrorAlertConfig log.AfterErrorConfiguration `yaml:"errorAlert"`
+	HeartbeatConfig  heartbeat.Config            `yaml:"heartbeat"`
 }
 
 type Defaults struct {

pkg/discovery/kubernetes/controller/selectpodhandler.go

@@ -364,7 +364,7 @@ func (c *Controller) makeConfigPerSource(s *source.Config, pod *corev1.Pod, lgc
 		}
 
 		// change the source name, add pod.Name-containerName as prefix, since there maybe multiple containers in pod
-		filesrc.Name = helper.GenTypePodSourceName(pod.Name, status.Name, filesrc.Name)
+		filesrc.Name = helper.GenTypePodSourceName(lgc.Namespace, pod.Namespace, pod.Name, status.Name, filesrc.Name)
 
 		// inject default pod metadata
 		if err := c.injectTypePodFields(c.config.DynamicContainerLog, filesrc, extra, pod, lgc, status.Name); err != nil {