Merge pull request #409 from kongfei605/logs_filter

add filter policy for pod log collecting

config/logs.go

@@ -28,6 +28,8 @@ type (
 		ScanPeriod            int                          `json:"scan_period" toml:"scan_period"`
 		FrameSize             int                          `json:"frame_size" toml:"frame_size"`
 		CollectContainerAll   bool                         `json:"collect_container_all" toml:"collect_container_all"`
+		ContainerInclude      []string                     `json:"container_include" toml:"container_include"`
+		ContainerExclude      []string                     `json:"container_exclude" toml:"container_exclude"`
 		GlobalProcessingRules []*logsconfig.ProcessingRule `json:"processing_rules" toml:"processing_rules"`
 		Items                 []*logsconfig.LogsConfig     `json:"items" toml:"items"`
 		KafkaConfig
@@ -88,3 +90,17 @@ func IsFeaturePresent(t string) bool {
 func GetContainerCollectAll() bool {
 	return Config.Logs.CollectContainerAll
 }
+
+func GetContainerIncludeList() []string {
+	if Config.Logs.ContainerInclude == nil {
+		return []string{}
+	}
+	return Config.Logs.ContainerInclude
+}
+
+func GetContainerExcludeList() []string {
+	if Config.Logs.ContainerExclude == nil {
+		return []string{}
+	}
+	return Config.Logs.ContainerExclude
+}

logs/input/kubernetes/launcher.go

@@ -32,6 +32,7 @@ const (
 	AnnotationRuleKey      = "categraf/logs.stdout.processing_rules"
 	AnnotationTopicKey     = "categraf/logs.stdout.topic"
 	AnnotationTagPrefixKey = "categraf/tags.prefix"
+	AnnotationCollectKey   = "categraf/logs.stdout.collect"
 )
 
 var (
@@ -254,6 +255,11 @@ func (l *Launcher) getSource(pod *kubelet.Pod, container kubelet.ContainerStatus
 		if !l.collectAll {
 			return nil, errCollectAllDisabled
 		}
+		if !(pod.Metadata.Annotations[AnnotationCollectKey] == "" ||
+			pod.Metadata.Annotations[AnnotationCollectKey] == "true") {
+			log.Printf("pod %s disable stdout collecting", pod.Metadata.Name)
+			return nil, errCollectAllDisabled
+		}
 		// The logs source is the short image name
 		logsSource := ""
 		shortImageName, err := l.getShortImageName(pod, container.Name)

logs/util/containers/filter.go

@@ -13,6 +13,8 @@ import (
 	"log"
 	"regexp"
 	"strings"
+
+	coreconfig "flashcat.cloud/categraf/config"
 )
 
 const (
@@ -60,6 +62,7 @@ const (
 	pauseContainerUpstream = `image:upstream/pause.*`
 	// - cdk/pause-amd64
 	pauseContainerCDK = `image:cdk/pause.*`
+	categrafContainer = `image:flashcatcloud/categraf.*`
 
 	// filter prefixes for inclusion/exclusion
 	imageFilterPrefix         = `image:`
@@ -200,8 +203,8 @@ func NewFilter(includeList, excludeList []string) (*Filter, error) {
 func newMetricFilterFromConfig() (*Filter, error) {
 	// We merge `container_include` and `container_include_metrics` as this filter
 	// is used by all core and python checks (so components sending metrics).
-	includeList := []string{}
-	excludeList := []string{}
+	includeList := coreconfig.GetContainerIncludeList()
+	excludeList := coreconfig.GetContainerExcludeList()
 
 	excludeList = append(excludeList,
 		pauseContainerGCR,
@@ -219,6 +222,7 @@ func newMetricFilterFromConfig() (*Filter, error) {
 		pauseContainerECR,
 		pauseContainerUpstream,
 		pauseContainerCDK,
+		categrafContainer,
 	)
 	return NewFilter(includeList, excludeList)
 }
@@ -228,15 +232,19 @@ func newMetricFilterFromConfig() (*Filter, error) {
 // It allows to filter metrics and logs separately
 // For use in autodiscovery.
 func NewAutodiscoveryFilter(filter FilterType) (*Filter, error) {
-	includeList := []string{}
-	excludeList := []string{}
+	includeList := coreconfig.GetContainerIncludeList()
+	excludeList := coreconfig.GetContainerExcludeList()
 	switch filter {
 	case GlobalFilter:
 		includeList = []string{}
-		excludeList = []string{"image:*.categraf.*"}
+		if len(excludeList) == 0 {
+			excludeList = append(excludeList, categrafContainer)
+		}
 	case LogsFilter:
 		includeList = []string{}
-		excludeList = []string{"image:.*categraf.*"}
+		if len(excludeList) == 0 {
+			excludeList = append(excludeList, categrafContainer)
+		}
 	}
 	return NewFilter(includeList, excludeList)
 }
@@ -267,7 +275,11 @@ func (cf Filter) IsExcluded(containerName, containerImage, podNamespace string)
 
 	// Check if excludeListed
 	for _, r := range cf.ImageExcludeList {
-		if r.MatchString(containerImage) {
+		match := r.MatchString(containerImage)
+		if coreconfig.Config.DebugMode {
+			log.Printf("D!, exclude item :%+v, container image:%s, %t\n", r, containerImage, match)
+		}
+		if match {
 			return true
 		}
 	}

logs/util/kubernetes/kubelet/kubelet.go

@@ -16,6 +16,7 @@ import (
 	"sync"
 	"time"
 
+	coreconfig "flashcat.cloud/categraf/config"
 	"flashcat.cloud/categraf/logs/errors"
 	"flashcat.cloud/categraf/logs/util/containers"
 	"flashcat.cloud/categraf/logs/util/containers/providers"
@@ -87,6 +88,10 @@ func NewKubeUtil() *KubeUtil {
 		podListCacheDuration: 5 * time.Second,
 		podUnmarshaller:      newPodUnmarshaller(),
 	}
+	filter, err := containers.NewAutodiscoveryFilter(containers.LogsFilter)
+	if err == nil {
+		ku.filter = filter
+	}
 
 	waitOnMissingContainer := 0
 	if waitOnMissingContainer > 0 {
@@ -211,7 +216,9 @@ func (ku *KubeUtil) GetLocalPodList(ctx context.Context) ([]*Pod, error) {
 			allContainers = append(allContainers, pod.Status.InitContainers...)
 			allContainers = append(allContainers, pod.Status.Containers...)
 			pod.Status.AllContainers = allContainers
-			tmpSlice = append(tmpSlice, pod)
+			if !ku.filterPod(pod) {
+				tmpSlice = append(tmpSlice, pod)
+			}
 		}
 	}
 	pods.Items = tmpSlice
@@ -222,6 +229,18 @@ func (ku *KubeUtil) GetLocalPodList(ctx context.Context) ([]*Pod, error) {
 	return pods.Items, nil
 }
 
+func (ku *KubeUtil) filterPod(pod *Pod) bool {
+	for _, c := range pod.Status.GetAllContainers() {
+		if coreconfig.Config.DebugMode {
+			log.Printf("D! container name:%s image:%s, ns:%s", c.Name, c.Image, pod.Metadata.Namespace)
+		}
+		if ku.filter.IsExcluded(c.Name, c.Image, pod.Metadata.Namespace) {
+			return true
+		}
+	}
+	return false
+}
+
 // ForceGetLocalPodList reset podList cache and call GetLocalPodList
 func (ku *KubeUtil) ForceGetLocalPodList(ctx context.Context) ([]*Pod, error) {
 	ResetCache()