-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
9 changed files
with
290 additions
and
59 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package kubernetes | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| "github.com/elastic/beats/libbeat/autodiscover/template" | ||
| ) | ||
|
|
||
| // Config for kubernetes autodiscover provider | ||
| type Config struct { | ||
| InCluster bool `config:"in_cluster"` | ||
| KubeConfig string `config:"kube_config"` | ||
| Host string `config:"host"` | ||
| Namespace string `config:"namespace"` | ||
| SyncPeriod time.Duration `config:"sync_period"` | ||
| CleanupTimeout time.Duration `config:"cleanup_timeout"` | ||
|
|
||
| IncludeLabels []string `config:"include_labels"` | ||
| ExcludeLabels []string `config:"exclude_labels"` | ||
| IncludeAnnotations []string `config:"include_annotations"` | ||
|
|
||
| Templates template.MapperSettings `config:"templates"` | ||
| } | ||
|
|
||
| func defaultConfig() *Config { | ||
| return &Config{ | ||
| InCluster: true, | ||
| SyncPeriod: 1 * time.Second, | ||
| } | ||
| } |
175 changes: 175 additions & 0 deletions
175
libbeat/autodiscover/providers/kubernetes/kubernetes.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,175 @@ | ||
| package kubernetes | ||
|
|
||
| import ( | ||
| "github.com/elastic/beats/libbeat/autodiscover" | ||
| "github.com/elastic/beats/libbeat/autodiscover/template" | ||
| "github.com/elastic/beats/libbeat/common" | ||
| "github.com/elastic/beats/libbeat/common/bus" | ||
| "github.com/elastic/beats/libbeat/common/kubernetes" | ||
| "github.com/elastic/beats/libbeat/logp" | ||
| ) | ||
|
|
||
| func init() { | ||
| autodiscover.ProviderRegistry.AddProvider("kubernetes", AutodiscoverBuilder) | ||
| } | ||
|
|
||
| // Provider implements autodiscover provider for docker containers | ||
| type Provider struct { | ||
| config *Config | ||
| bus bus.Bus | ||
| watcher kubernetes.Watcher | ||
| metagen kubernetes.MetaGenerator | ||
| templates *template.Mapper | ||
| stop chan interface{} | ||
| startListener bus.Listener | ||
| stopListener bus.Listener | ||
| updateListener bus.Listener | ||
| } | ||
|
|
||
| // AutodiscoverBuilder builds and returns an autodiscover provider | ||
| func AutodiscoverBuilder(bus bus.Bus, c *common.Config) (autodiscover.Provider, error) { | ||
| config := defaultConfig() | ||
| err := c.Unpack(&config) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| mapper, err := template.NewConfigMapper(config.Templates) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| client, err := kubernetes.GetKubernetesClient(config.InCluster, config.KubeConfig) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| metagen := kubernetes.NewMetaGenerator(config.IncludeAnnotations, config.IncludeLabels, config.ExcludeLabels) | ||
|
|
||
| config.Host = kubernetes.DiscoverKubernetesNode(config.Host, client) | ||
| watcher := kubernetes.NewWatcher(client.CoreV1(), config.SyncPeriod, config.CleanupTimeout, config.Host) | ||
|
|
||
| start := watcher.ListenStart() | ||
| stop := watcher.ListenStop() | ||
| update := watcher.ListenUpdate() | ||
|
|
||
| if err := watcher.Start(); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &Provider{ | ||
| config: config, | ||
| bus: bus, | ||
| templates: mapper, | ||
| metagen: metagen, | ||
| watcher: watcher, | ||
| stop: make(chan interface{}), | ||
| startListener: start, | ||
| stopListener: stop, | ||
| updateListener: update, | ||
| }, nil | ||
| } | ||
|
|
||
| func (p *Provider) Start() { | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-p.stop: | ||
| p.startListener.Stop() | ||
| p.stopListener.Stop() | ||
| return | ||
|
|
||
| case event := <-p.startListener.Events(): | ||
| p.emit(event, "start") | ||
|
|
||
| case event := <-p.stopListener.Events(): | ||
| p.emit(event, "stop") | ||
|
|
||
| case event := <-p.updateListener.Events(): | ||
| //On updates, first send a stop signal followed by a start signal to simulate a restart | ||
| p.emit(event, "stop") | ||
| p.emit(event, "start") | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| func (p *Provider) emit(event bus.Event, flag string) { | ||
| pod, ok := event["pod"].(*kubernetes.Pod) | ||
| if !ok { | ||
| logp.Err("Couldn't get a pod from watcher event") | ||
| return | ||
| } | ||
|
|
||
| host := pod.Status.PodIP | ||
|
|
||
| // Emit pod container IDs | ||
| for _, c := range append(pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses...) { | ||
| cmeta := common.MapStr{ | ||
| "id": c.GetContainerID(), | ||
| "name": c.Name, | ||
| "image": c.Image, | ||
| } | ||
|
|
||
| // Metadata appended to each event | ||
| meta := p.metagen.ContainerMetadata(pod, c.Name) | ||
|
|
||
| // Information that can be used in discovering a workload | ||
| kubemeta := meta.Clone() | ||
| kubemeta["container"] = cmeta | ||
|
|
||
| // Emit container info | ||
| p.publish(bus.Event{ | ||
| flag: true, | ||
| "host": host, | ||
| "kubernetes": kubemeta, | ||
| "meta": common.MapStr{ | ||
| "kubernetes": meta, | ||
| }, | ||
| }) | ||
| } | ||
|
|
||
| // Emit pod ports | ||
| for _, c := range pod.Spec.Containers { | ||
| cmeta := common.MapStr{ | ||
| "name": c.Name, | ||
| "image": c.Image, | ||
| } | ||
|
|
||
| // Metadata appended to each event | ||
| meta := p.metagen.ContainerMetadata(pod, c.Name) | ||
|
|
||
| // Information that can be used in discovering a workload | ||
| kubemeta := meta.Clone() | ||
| kubemeta["container"] = cmeta | ||
|
|
||
| for _, port := range c.Ports { | ||
| event := bus.Event{ | ||
| flag: true, | ||
| "host": host, | ||
| "port": port.ContainerPort, | ||
| "kubernetes": kubemeta, | ||
| "meta": common.MapStr{ | ||
| "kubernetes": meta, | ||
| }, | ||
| } | ||
| p.publish(event) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (p *Provider) publish(event bus.Event) { | ||
| // Try to match a config | ||
| if config := p.templates.GetConfig(event); config != nil { | ||
| event["config"] = config | ||
| } | ||
| p.bus.Publish(event) | ||
| } | ||
|
|
||
| func (p *Provider) Stop() { | ||
| close(p.stop) | ||
| } | ||
|
|
||
| func (p *Provider) String() string { | ||
| return "kubernetes" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package kubernetes | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "os" | ||
|
|
||
| "github.com/ericchiang/k8s" | ||
| "github.com/ghodss/yaml" | ||
|
|
||
| "github.com/elastic/beats/libbeat/logp" | ||
| ) | ||
|
|
||
| func GetKubernetesClient(in_cluster bool, kube_config string) (client *k8s.Client, err error) { | ||
| if in_cluster == true { | ||
| client, err = k8s.NewInClusterClient() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("Unable to get in cluster configuration: %v", err) | ||
| } | ||
| } else { | ||
| data, err := ioutil.ReadFile(kube_config) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("read kubeconfig: %v", err) | ||
| } | ||
|
|
||
| // Unmarshal YAML into a Kubernetes config object. | ||
| var config k8s.Config | ||
| if err = yaml.Unmarshal(data, &config); err != nil { | ||
| return nil, fmt.Errorf("unmarshal kubeconfig: %v", err) | ||
| } | ||
| client, err = k8s.NewClient(&config) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| } | ||
|
|
||
| return client, nil | ||
| } | ||
|
|
||
| func DiscoverKubernetesNode(host string, client *k8s.Client) string { | ||
| ctx := context.Background() | ||
| if host == "" { | ||
| podName := os.Getenv("HOSTNAME") | ||
| logp.Info("Using pod name %s and namespace %s", podName, client.Namespace) | ||
| if podName == "localhost" { | ||
| host = "localhost" | ||
| } else { | ||
| pod, error := client.CoreV1().GetPod(ctx, podName, client.Namespace) | ||
| if error != nil { | ||
| logp.Err("Querying for pod failed with error: ", error.Error()) | ||
| logp.Info("Unable to find pod, setting host to localhost") | ||
| host = "localhost" | ||
| } else { | ||
| host = pod.Spec.GetNodeName() | ||
| } | ||
|
|
||
| } | ||
| } | ||
|
|
||
| return host | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.