2 Star 1 Fork 0

李玮 / trireme-lib

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
monitor.go 3.81 KB
一键复制 编辑 原始数据 按行查看 历史
李玮 提交于 2020-01-29 13:23 . v1
package kubernetesmonitor
import (
"context"
"fmt"
"git.cloud.top/DSec/trireme-lib/collector"
"git.cloud.top/DSec/trireme-lib/monitor/config"
"git.cloud.top/DSec/trireme-lib/monitor/extractors"
dockermonitor "git.cloud.top/DSec/trireme-lib/monitor/internal/docker"
"git.cloud.top/DSec/trireme-lib/monitor/registerer"
"go.uber.org/zap"
"k8s.io/client-go/kubernetes"
kubecache "k8s.io/client-go/tools/cache"
)
// KubernetesMonitor implements a monitor that sends pod events upstream
// It is implemented as a filter on the standard DockerMonitor.
// It gets all the PU events from the DockerMonitor and if the container is the POD container from Kubernetes,
// It connects to the Kubernetes API and adds the tags that are coming from Kuberntes that cannot be found
type KubernetesMonitor struct {
dockerMonitor *dockermonitor.DockerMonitor
kubeClient kubernetes.Interface
localNode string
handlers *config.ProcessorConfig
cache *cache
kubernetesExtractor extractors.KubernetesMetadataExtractorType
podStore kubecache.Store
podController kubecache.Controller
podControllerStop chan struct{}
enableHostPods bool
}
// New returns a new kubernetes monitor.
func New() *KubernetesMonitor {
kubeMonitor := &KubernetesMonitor{}
kubeMonitor.cache = newCache()
return kubeMonitor
}
// SetupConfig provides a configuration to implmentations. Every implmentation
// can have its own config type.
func (m *KubernetesMonitor) SetupConfig(registerer registerer.Registerer, cfg interface{}) error {
defaultConfig := DefaultConfig()
if cfg == nil {
cfg = defaultConfig
}
kubernetesconfig, ok := cfg.(*Config)
if !ok {
return fmt.Errorf("Invalid configuration specified")
}
kubernetesconfig = SetupDefaultConfig(kubernetesconfig)
processorConfig := &config.ProcessorConfig{
Policy: m,
Collector: collector.NewDefaultCollector(),
}
// As the Kubernetes monitor depends on the DockerMonitor, we setup the Docker monitor first
dockerMon := dockermonitor.New()
dockerMon.SetupHandlers(processorConfig)
dockerConfig := dockermonitor.DefaultConfig()
dockerConfig.EventMetadataExtractor = kubernetesconfig.DockerExtractor
// we use the defaultconfig for now
if err := dockerMon.SetupConfig(nil, dockerConfig); err != nil {
return fmt.Errorf("docker monitor instantiation error: %s", err.Error())
}
m.dockerMonitor = dockerMon
// Setting up Kubernetes
m.localNode = kubernetesconfig.Nodename
kubeClient, err := NewKubeClient(kubernetesconfig.Kubeconfig)
if err != nil {
return fmt.Errorf("kubernetes client instantiation error: %s", err.Error())
}
m.kubeClient = kubeClient
m.enableHostPods = kubernetesconfig.EnableHostPods
m.kubernetesExtractor = kubernetesconfig.KubernetesExtractor
m.podStore, m.podController = m.CreateLocalPodController("",
m.addPod,
m.deletePod,
m.updatePod)
m.podControllerStop = make(chan struct{})
zap.L().Warn("Using deprecated Kubernetes Monitor")
return nil
}
// Run starts the monitor.
func (m *KubernetesMonitor) Run(ctx context.Context) error {
if m.kubeClient == nil {
return fmt.Errorf("kubernetes client is not initialized correctly")
}
// TODO. Give directly the channel to the Kubernetes library
go m.podController.Run(ctx.Done())
initialPodSync := make(chan struct{})
go hasSynced(initialPodSync, m.podController)
<-initialPodSync
return m.dockerMonitor.Run(ctx)
}
// SetupHandlers sets up handlers for monitors to invoke for various events such as
// processing unit events and synchronization events. This will be called before Start()
// by the consumer of the monitor
func (m *KubernetesMonitor) SetupHandlers(c *config.ProcessorConfig) {
m.handlers = c
}
// Resync requests to the monitor to do a resync.
func (m *KubernetesMonitor) Resync(ctx context.Context) error {
return m.dockerMonitor.Resync(ctx)
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/emmoblin/trireme-lib.git
git@gitee.com:emmoblin/trireme-lib.git
emmoblin
trireme-lib
trireme-lib
7726874a2b9a

搜索帮助

344bd9b3 5694891 D2dac590 5694891