feat: global klog standardization

agent/agent.go

@@ -2,7 +2,8 @@ package agent
 
 import (
 	"errors"
-	"log"
+
+	"k8s.io/klog/v2"
 )
 
 type Agent struct {
@@ -35,38 +36,38 @@ func NewAgent() (*Agent, error) {
 }
 
 func (a *Agent) Start() {
-	log.Println("I! agent starting")
+	klog.InfoS("agent starting")
 	for _, agent := range a.agents {
 		if agent == nil {
 			continue
 		}
 		if err := agent.Start(); err != nil {
-			log.Printf("E! start [%T] err: [%+v]", agent, err)
+			klog.ErrorS(err, "start agent module failed", "module", agent)
 		} else {
-			log.Printf("I! [%T] started", agent)
+			klog.InfoS("agent module started", "module", agent)
 		}
 	}
-	log.Println("I! agent started")
+	klog.InfoS("agent started")
 }
 
 func (a *Agent) Stop() {
-	log.Println("I! agent stopping")
+	klog.InfoS("agent stopping")
 	for _, agent := range a.agents {
 		if agent == nil {
 			continue
 		}
 		if err := agent.Stop(); err != nil {
-			log.Printf("E! stop [%T] err: [%+v]", agent, err)
+			klog.ErrorS(err, "stop agent module failed", "module", agent)
 		} else {
-			log.Printf("I! [%T] stopped", agent)
+			klog.InfoS("agent module stopped", "module", agent)
 		}
 	}
-	log.Println("I! agent stopped")
+	klog.InfoS("agent stopped")
 }
 
 func (a *Agent) Reload() {
-	log.Println("I! agent reloading")
+	klog.InfoS("agent reloading")
 	a.Stop()
 	a.Start()
-	log.Println("I! agent reloaded")
+	klog.InfoS("agent reloaded")
 }

agent/ibex_agent.go

@@ -3,10 +3,9 @@
 package agent
 
 import (
-	"log"
-
 	coreconfig "flashcat.cloud/categraf/config"
 	"flashcat.cloud/categraf/ibex"
+	"k8s.io/klog/v2"
 )
 
 type IbexAgent struct {
@@ -16,7 +15,7 @@ func NewIbexAgent() AgentModule {
 	if coreconfig.Config == nil ||
 		coreconfig.Config.Ibex == nil ||
 		!coreconfig.Config.Ibex.Enable {
-		log.Println("I! ibex agent disabled!")
+		klog.Info("ibex agent disabled")
 		return nil
 	}
 	if coreconfig.Config.Ibex.MetaDir == "" {

agent/install/service_linux.go

@@ -3,14 +3,14 @@ package install
 import (
 	"bytes"
 	"fmt"
-	"log"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/kardianos/service"
+	"k8s.io/klog/v2"
 
 	"flashcat.cloud/categraf/pkg/cmdx"
 )
@@ -167,7 +167,7 @@ func isSystemd() bool {
 
 	err, timeout := cmdx.RunTimeout(cmd, time.Second*2)
 	if timeout {
-		log.Printf("E! run command: %s timeout", cmd)
+		klog.Warningf("run command timeout: %s", cmd)
 		return false
 	}
 
@@ -234,7 +234,7 @@ func ServiceConfig(userMode bool) *service.Config {
 			cfg.WorkingDirectory = filepath.Dir(ov)
 		}
 	} else {
-		log.Println("E! get exeutable path error:", err)
+		klog.ErrorS(err, "get executable path error")
 	}
 	cfg.Arguments = []string{"-configs", filepath.Dir(ov) + "/conf"}
 	if userMode {

agent/logs_agent.go

@@ -11,7 +11,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
 	"os"
 	"time"
 
@@ -31,6 +30,7 @@ import (
 	coreconfig "flashcat.cloud/categraf/config"
 	logsconfig "flashcat.cloud/categraf/config/logs"
 	logService "flashcat.cloud/categraf/logs/service"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -71,20 +71,20 @@ func NewLogsAgent() AgentModule {
 	if err != nil {
 		message := fmt.Sprintf("Invalid endpoints: %v", err)
 		status.AddGlobalError("invalid endpoints", message)
-		log.Println("E!", errors.New(message))
+		klog.ErrorS(errors.New(message), "invalid endpoints")
 		return nil
 	}
 	processingRules, err := GlobalProcessingRules()
 	if err != nil {
 		message := fmt.Sprintf("Invalid processing rules: %v", err)
 		status.AddGlobalError(invalidProcessingRules, message)
-		log.Println("E!", errors.New(message))
+		klog.ErrorS(errors.New(message), "invalid processing rules")
 		return nil
 	}
 
 	sources := logsconfig.NewLogSources()
 	services := logService.NewServices()
-	log.Println("I! Starting logs-agent...")
+	klog.Info("starting logs-agent")
 
 	// setup the auditor
 	// We pass the health handle to the auditor because it's the end of the pipeline and the most
@@ -120,7 +120,7 @@ func NewLogsAgent() AgentModule {
 		journald.NewLauncher(sources, pipelineProvider, auditor),
 	}
 	if coreconfig.EnableCollectContainer() {
-		log.Println("collect docker logs...")
+		klog.Info("collect docker logs")
 		inputs = append(inputs, container.NewLauncher(containerLaunchables))
 	}
 
@@ -142,7 +142,7 @@ func (la *LogsAgent) Start() error {
 	if coreconfig.EnableCollectContainer() {
 		// collect container all
 		if util.Debug() {
-			log.Println("Adding ContainerCollectAll source to the Logs Agent")
+			klog.V(1).Info("adding ContainerCollectAll source to the Logs Agent")
 		}
 		kubesource := logsconfig.NewLogSource(logsconfig.ContainerCollectAll,
 			&logsconfig.LogsConfig{
@@ -161,7 +161,7 @@ func (la *LogsAgent) Start() error {
 		}
 		source := logsconfig.NewLogSource(c.Name, c)
 		if err := c.Validate(); err != nil {
-			log.Println("W! Invalid logs configuration:", err)
+			klog.Warningf("invalid logs configuration: %v", err)
 			source.Status.Error(err)
 			continue
 		}
@@ -214,7 +214,7 @@ func (a *LogsAgent) Stop() error {
 	select {
 	case <-c:
 	case <-time.After(timeout):
-		log.Println("I! Timed out when stopping logs-agent, forcing it to stop now")
+		klog.Info("timed out when stopping logs-agent, forcing it to stop now")
 		// We force all destinations to read/flush all the messages they get without
 		// trying to write to the network.
 		a.destinationsCtx.Stop()
@@ -226,7 +226,7 @@ func (a *LogsAgent) Stop() error {
 		select {
 		case <-c:
 		case <-timeout.C:
-			log.Println("W! Force close of the Logs LogsAgent, dumping the Go routines.")
+			klog.Warning("force close of the logs agent, dumping the Go routines")
 		}
 	}
 	return nil

agent/metrics_agent.go

@@ -2,7 +2,6 @@ package agent
 
 import (
 	"errors"
-	"log"
 	"strings"
 	"sync"
 
@@ -104,6 +103,7 @@ import (
 	_ "flashcat.cloud/categraf/inputs/x509_cert"
 	_ "flashcat.cloud/categraf/inputs/xskyapi"
 	_ "flashcat.cloud/categraf/inputs/zookeeper"
+	"k8s.io/klog/v2"
 )
 
 type MetricsAgent struct {
@@ -167,7 +167,7 @@ func NewMetricsAgent() AgentModule {
 
 	provider, err := inputs.NewProvider(c, agent)
 	if err != nil {
-		log.Println("E! init metrics agent error: ", err)
+		klog.ErrorS(err, "init metrics agent error")
 		return nil
 	}
 	agent.InputProviders = provider
@@ -196,7 +196,7 @@ func (ma *MetricsAgent) Start() error {
 
 func (ma *MetricsAgent) start(idx int) error {
 	if _, err := ma.InputProviders[idx].LoadConfig(); err != nil {
-		log.Println("E! input provider load config get err: ", err)
+		klog.ErrorS(err, "input provider load config get err")
 	}
 	ma.InputProviders[idx].StartReloader()
 
@@ -206,7 +206,7 @@ func (ma *MetricsAgent) start(idx int) error {
 	}
 
 	if len(names) == 0 {
-		log.Println("I! no inputs")
+		klog.InfoS("no inputs")
 		return nil
 	}
 
@@ -218,7 +218,7 @@ func (ma *MetricsAgent) start(idx int) error {
 
 		configs, err := ma.InputProviders[idx].GetInputConfig(name)
 		if err != nil {
-			log.Println("E! failed to get configuration of plugin:", name, "error:", err)
+			klog.ErrorS(err, "failed to get configuration of plugin", "plugin", name)
 			continue
 		}
 
@@ -249,7 +249,7 @@ func (ma *MetricsAgent) RegisterInput(name string, configs []cfg.ConfigWithForma
 
 	creator, has := inputs.InputCreators[inputKey]
 	if !has {
-		log.Println("E! input:", name, "not supported")
+		klog.Warningf("input %s not supported", name)
 		return
 	}
 
@@ -260,12 +260,12 @@ func (ma *MetricsAgent) RegisterInput(name string, configs []cfg.ConfigWithForma
 		}
 	}
 	if idx == -1 {
-		log.Println("E! input provider:", typ, "not found")
+		klog.Warningf("input provider %s not found", typ)
 		// hint and panic next line
 	}
 	newInputs, err := ma.InputProviders[idx].LoadInputConfig(configs, creator())
 	if err != nil {
-		log.Println("E! failed to load configuration of plugin:", name, "error:", err)
+		klog.ErrorS(err, "failed to load configuration of plugin", "plugin", name)
 		return
 	}
 
@@ -276,19 +276,18 @@ func (ma *MetricsAgent) RegisterInput(name string, configs []cfg.ConfigWithForma
 
 func (ma *MetricsAgent) inputGo(name string, sum string, input inputs.Input) {
 	var err error
+	inputLogger := klog.Background().WithValues(metricsAgentInputLoggerValues(name, sum)...)
 	if err = input.InitInternalConfig(); err != nil {
-		log.Println("E! failed to init input:", name, "error:", err)
+		klog.ErrorS(err, "failed to init input", "input", name)
 		return
 	}
 
-	if err = inputs.MayInit(input); err != nil {
+	if err = inputs.MayInit(input, inputLogger); err != nil {
 		if !errors.Is(err, types.ErrInstancesEmpty) {
-			log.Println("E! failed to init input:", name, "error:", err)
+			klog.ErrorS(err, "failed to init input", "input", name)
 		} else {
-			if config.Config.DebugMode {
-				_, inputKey := inputs.ParseInputName(name)
-				log.Println("W! no instances for input: ", inputKey)
-			}
+			_, inputKey := inputs.ParseInputName(name)
+			klog.V(1).InfoS("no instances for input", "input", inputKey)
 		}
 		return
 	}
@@ -298,13 +297,14 @@ func (ma *MetricsAgent) inputGo(name string, sum string, input inputs.Input) {
 		empty := true
 		for i := 0; i < len(instances); i++ {
 			if err := instances[i].InitInternalConfig(); err != nil {
-				log.Println("E! failed to init input:", name, "error:", err)
+				klog.ErrorS(err, "failed to init input", "input", name)
 				continue
 			}
 
-			if err := inputs.MayInit(instances[i]); err != nil {
+			instanceLogger := inputLogger.WithValues(metricsAgentInstanceLoggerValues(i, instances[i].GetLabels())...)
+			if err := inputs.MayInit(instances[i], instanceLogger); err != nil {
 				if !errors.Is(err, types.ErrInstancesEmpty) {
-					log.Println("E! failed to init input:", name, "error:", err)
+					klog.ErrorS(err, "failed to init input", "input", name)
 				}
 				continue
 			}
@@ -313,18 +313,16 @@ func (ma *MetricsAgent) inputGo(name string, sum string, input inputs.Input) {
 		}
 
 		if empty {
-			if config.Config.DebugMode {
-				_, inputKey := inputs.ParseInputName(name)
-				log.Printf("W! no instances for input:%s", inputKey)
-			}
+			_, inputKey := inputs.ParseInputName(name)
+			klog.V(1).InfoS("no instances for input", "input", inputKey)
 			return
 		}
 	}
 
 	reader := newInputReader(name, input)
 	go reader.startInput()
 	ma.InputReaders.Add(name, sum, reader)
-	log.Println("I! input:", name, "started")
+	klog.InfoS("input started", "input", name)
 }
 
 func (ma *MetricsAgent) DeregisterInput(name string, sum string) {
@@ -335,9 +333,9 @@ func (ma *MetricsAgent) DeregisterInput(name string, sum string) {
 			}
 		}
 		ma.InputReaders.Del(name, sum)
-		log.Printf("I! input: %s[checksum:%s] stopped", name, sum)
+		klog.InfoS("input stopped", "input", name, "checksum", sum)
 	} else {
-		log.Printf("W! dereigster input name [%s] not found", name)
+		klog.Warningf("deregister input name [%s] not found", name)
 	}
 }
 
@@ -352,3 +350,21 @@ func parseFilter(filterStr string) map[string]struct{} {
 	}
 	return filtermap
 }
+
+func metricsAgentInputLoggerValues(name, sum string) []interface{} {
+	_, plugin := inputs.ParseInputName(name)
+	return []interface{}{
+		"component", "inputs",
+		"input", name,
+		"plugin", plugin,
+		"checksum", sum,
+	}
+}
+
+func metricsAgentInstanceLoggerValues(index int, labels map[string]string) []interface{} {
+	values := []interface{}{"instance_index", index}
+	if target, ok := labels["target"]; ok && target != "" {
+		values = append(values, "instance_target", target)
+	}
+	return values
+}