New `kubernetes_system_events` source

[binarylogic]

Kubernetes exposes an event stream for system events. I'd like to collect these as an option and run them through the pipeline like any other log event.

Prior Art

• https://github.com/heptiolabs/eventrouter
• https://github.com/kubernetes-retired/heapster
• https://docs.datadoghq.com/agent/kubernetes/event_collection/
• https://docs.newrelic.com/docs/integrations/kubernetes-integration/kubernetes-events/new-relic-kubernetes-events-integration

Helpful info

• https://www.bluematador.com/blog/kubernetes-events-explained
• https://kubernetes.io/docs/tasks/debug-application-cluster/debug-application-introspection/
• https://github.com/aws/containers-roadmap/issues/566

[ktff]

Description

A dedicated source for collecting Kubernetes events.

https://github.com/heptiolabs/eventrouter is a good example.

Behavior

Source will collect events from it's local Kubernetes cluster API server.

Configuration

[sources.my_source_id]
  # REQUIRED - General
  type = "kubernetes_events" # must be: "kubernetes_events"

Deployment

For deploying in Kubernetes cluster, a vector.yaml file is necessary. It would be similar to the one for kubernetes_source, with few differences.

It would be a Deployment with one replica. Whose installation, configuration, and reconfiguration would be done with

kubectl apply -f yaml_path

Which is the same way as kubernetes_source in #260.

Implementation

Source will register to watch for events on Kubernetes API of local cluster on /api/v1/events.

kube crate would be used, which is currently only used during testing.

Events from before Vector has been started will be collected on best effort basis. And after, all events will be collected. This behavior will provide some coverage of events happening between Vector restarts. Which mostly happens when a node goes down.

Event fields

From all fields supplied by Kubernetes API, only some technical fields will be filtered out while also preserving naming schema as much as possible to be compatible with Kubernetes documentation.

The following would be present in the emitted events:

• action
• timestamp
• kind
• message
• reason
• type
• cluster_name
• reporting_component
• reporting_instance
• source.component
• source.host
• object.field_path
• object.kind
• object.name
• object.namespace
• object.uid - corresponds to object_uid in #1249

With optional:

• related.field_path
• related.kind
• related.name
• related.namespace
• related.uid - corresponds to object_uid in #1249

---

Extensions

Election

To allow having multiple kubernetes_events sources in a single cluster, an election process is necessary. With it, only one kubernetes_events source could be elected to collect events from Kubernetes API.

With this feature, kubernetes_events_source could be added to Vector configuration alongside of kubernetes_source. With that, separate toml/yaml/installation/etc. for kubernetes_events_source can be avoided.

To avoid P2P communication, properties of Kubernetes API objects could be used to implement the election. https://kubernetes.io/blog/2016/01/simple-leader-election-with-kubernetes/ is an example of it.

This would also add more robustness against node failure.

EDIT: The same election process can be used to avoid pulling all backlog for each new elected leader, but just events that happened from the time past leader last reported in.

DataDog has election process. Although the implementation is unknown.

As kubernetes_source option

With Election extension in place, kubernetes_source could gain an option to collect Kubernetes events, which would in the background spin this kubernetes_events_source.

Should this be a separate source or an option on the kubernetes source?

@binarylogic with this, we would have both for a price of one, because something like this source would have to be implemented in either way.

Filtering

• namespace - it's a good way of filtering when there is so many sources of events. And also, Kuberentes API supports it quite naturally.
• type - Normal,Warning, new types could be added in the future

External cluster

Source could be extended to accept optional credentials and address to external Kubernetes API server. That way Vector could collect Kubernetes events from outside the cluster.

This is especially useful for more reliable monitoring of cluster, as it doesn't depend on it or it's nodes.

Also, events could be collected from more than one cluster.

Centralized collection of logs

Using the same API, logs from all containers could be collected. Which use is greatly expanded with External cluster extension.

Although, such way of collecting logs is only suitable for smaller clusters.

---

Alternatives

Main problem with proposed implementation is the necessity for additional Vector/toml/yaml/source. Alternatives mainly tackle this issue.

There is one plausible alternative:

• Scrap local nodes system logs from files and journald.
  • It's unclear if all events could be collected like this. Probably not possible to collect events from system components without additional api server configuration
  • It's implementation dependent
  • Would collect more noise, but there is a good chance that we could deal with this

---

Other

• The need for separate yaml file / installation could be mitigated if added to supplied kubernetes_source yaml file.

[MOZGIII]

Closing this in favor of https://github.com/timberio/vector/issues/1424. Audit is pretty much the same data, but with fine-grained access control and less security risks. On the other hand, it needs more configuration, which we'll handle transparently out of the box with the Helm Chart.

[tyrken]

Can I challenge you a bit on this @MOZGIII - I don't see the very useful k8s events you get from kubectl get events in the audit data, event when searching for it. Admittedly I'm using EKS which as a managed services is limited in how it exposes audit/master logs, though I have activated them to a least a basic degree. I am double checking now if there's more I can enable , but the plain output of k get events is very clear & good with a high signal-to-noise level, unlike the audit data I can see so far.

This is very useful data - e.g. see https://github.com/salesforce/sloop which I'll probably have to install now, and is needed for full application visibility.

[ottramst]

Expanding on @tyrken's comment - our organization is moving away from Beats, because of the different issues we have seen with them so far and it would be greatly beneficial to have a separate source of Kubernetes events like Metricbeat does. https://www.elastic.co/guide/en/beats/metricbeat/current/metricbeat-metricset-kubernetes-event.html

Likely until there is some conclusion on this topic we need to keep running Metricbeat nevertheless. We are also running on EKS.

[spencergilbert]

Closing this in favor of #1424. Audit is pretty much the same data, but with fine-grained access control and less security risks. On the other hand, it needs more configuration, which we'll handle transparently out of the box with the Helm Chart.

I agree that audit logs and "kube events" are used differently (at least in my experience), and I've also not seen the same information across both streams. I think this issue is still valuable to have open.

[ondrejmo]

Please correct me if I am missing something, but I believe it's possible to relatively easily implement this with the current features. This is an example from my environment, so it's got some specifics.

sources:
  kube__events:
    type: http_client
    endpoint: https://kubernetes.default.svc:443/api/v1/events
    scrape_interval_secs: 15
    decoding: 
      codec: json
    auth:
      strategy: bearer
      token: "${KUBE_TOKEN}" # needs the Secret from the RBAC bellow as envvar
    tls:
      ca_file: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    headers:
      Accept:
        - application/json

transforms:
  kube__events_split:
    type: remap
    inputs: [ kube__events ]
    source: |-
      . = array!(.items)

      # no del()

  kube__events_dedupe:
    type: dedupe
    inputs: [ kube__events_split ]
    fields:
      match:
        - metadata.uid

  kube__events_remap:
    type: remap
    inputs: [ kube__events_dedupe ]
    source: |-
      .@topic = "kube.events"
      .timestamp = format_timestamp!(now(), format: "%Y-%m-%dT%H:%M:%S%.6fZ")

      # no del()

A ServiceAccount with some RBAC is needed by the Vector.dev instance.

---

apiVersion: v1
kind: ServiceAccount
metadata:
  name: aggregator

---

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: aggregator-kube-events

rules:
  - apiGroups: 
      - "*"
    resources: 
      - events
    verbs: 
      - get
      - list

---

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: aggregator-kube-events

subjects:
  - kind: ServiceAccount
    name: aggregator
    namespace: monitoring
roleRef:
  kind: ClusterRole
  name: aggregator-kube-events

---

apiVersion: v1
kind: Secret
metadata:
  name: aggregator
  annotations:
    kubernetes.io/service-account.name: aggregator

type: kubernetes.io/service-account-token

[camaeel]

IMHO querying all events in the cluster every 15 secs might create significant load on the apiservers.

[ondrejmo]

I suppose it might, thought in my (one master + one agent) cluster it seems ok.

[tai-nd]

Please correct me if I am missing something, but I believe it's possible to relatively easily implement this with the current features. This is an example from my environment, so it's got some specifics.

I think the problem is that Vector is usually deployed as DaemonSet in Kubernetes, so you have multiple Vector instances gather the same events. To solve that we need sharding or election, or just write/use a custom exporter (like your standalone agent).

[air3ijai]

Now we cover that with a Fluentbit - a single replica deployment.

[ondrejmo]

Please correct me if I am missing something, but I believe it's possible to relatively easily implement this with the current features. This is an example from my environment, so it's got some specifics.

I think the problem is that Vector is usually deployed as DaemonSet in Kubernetes, so you have multiple Vector instances gather the same events. To solve that we need sharding or election, or just write/use a custom exporter (like your standalone agent).

Yes, that's true, but this can be corrected by the deduplication stage in the example, if you either put it on a central aggregator or only scrape the api by the aggregator itself. Having one aggregator is also desirable for chunked writes to e.g. S3 storage.

[dm3ch]

Please correct me if I am missing something, but I believe it's possible to relatively easily implement this with the current features. This is an example from my environment, so it's got some specifics.

I think the problem is that Vector is usually deployed as DaemonSet in Kubernetes, so you have multiple Vector instances gather the same events. To solve that we need sharding or election, or just write/use a custom exporter (like your standalone agent).

I see 3 different options how to solve the problem with events duplication:

1. Collect events using separate Vector deployment (with Deployment type), in this case, duplication wouldn't occur.
2. Add a filtering stage to the ingestion pipeline, which would be effective for filtering duplicated events
3. Add some lease-acquiring logic to kuberenetes_event itself. Vector with kubernetes_event enabled would have access to Kubernetes API, so it would be possible to utilize Kubernetes lease object for that (https://kubernetes.io/docs/concepts/architecture/leases/)

I personally believe that 3rd option is the best, due to the following:

• It wouldn't require separate deployment.
• It wouldn't generate duplicated logs at all resulting in eliminating the need to have a separate filtering stage in ingestion pipeline. I believe it's better not to produce duplicated events instead of filtering them.
• It would be more fault-tolerant. If a current instance of vector collecting events would die, another one would be able to acquire a lease after lease timeout. Maybe it would be even possible to introduce a logic to re-read events starting from last lease renew time, to prevent situation where some of events wouldn't be streamed (but it can introduce event duplication only in this period of time).

P.S. But as far as I understand current problem is not in duplication, but in the fact, that no one has started development yet.

[orf]

For those interested, below is a working helm values file for this. Replace XXX placeholders with suitable values:

nameOverride: ""

fullnameOverride: ""

role: "Aggregator"

replicas: 1

service:
  enabled: false

env:
  - name: KUBE_TOKEN
    valueFrom:
      secretKeyRef:
        name: vector-token
        key: token

serviceAccount:
  name: vector-events
  create: true
  annotations:
    eks.amazonaws.com/role-arn: XXX

customConfig:
  data_dir: "/var/lib/vector"
  sources:
    kube_events:
      type: http_client
      endpoint: https://kubernetes.default.svc:443/api/v1/events
      query:
        fieldSelector: [XXX]
        namespace: [XXX]

      scrape_interval_secs: 5
      decoding:
        codec: json
      auth:
        strategy: bearer
        token: "${KUBE_TOKEN:?}"
      tls:
        ca_file: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
      headers:
        Accept:
          - application/json

  transforms:
    unnest:
      type: remap
      inputs: [ kube_events ]
      source: |
        . = unnest!(.items)

    normalized_events:
      type: remap
      inputs: [ unnest ]
      source: |
        . = {
          "run_id": null,
          "node_name": .items.reportingInstance,
          "pod_name": .items.involvedObject.name,
          "timestamp": .items.lastTimestamp,
          "message": .items.message,
        }

    deduped_events:
      type: dedupe
      inputs:
        - normalized_events
      fields:
        match: ["node_name", "pod_name", "message", "timestamp" ]


  sinks:
    s3_events_test:
      inputs:
        - "deduped_events"
      type: "aws_s3"
      region: XXX
      bucket: XXX
      key_prefix: 'logs/node_name={{ "{{" }} .node_name }}/events-'
      compression: "zstd"
      filename_extension: "json.zstd"
      framing:
        method: "newline_delimited"
      encoding:
        codec: "json"
      batch:
        timeout_secs: 300


extraObjects:
  - apiVersion: rbac.authorization.k8s.io/v1
    kind: ClusterRole
    metadata:
      name: vector-kube-events

    rules:
      - apiGroups:
          - "*"
        resources:
          - events
        verbs:
          - get
          - list

  - apiVersion: rbac.authorization.k8s.io/v1
    kind: ClusterRoleBinding
    metadata:
      name: vector-kube-events

    subjects:
      - kind: ServiceAccount
        name: vector-events
        namespace: vector
    roleRef:
      kind: ClusterRole
      name: vector-kube-events

  - apiVersion: v1
    kind: Secret
    metadata:
      name: vector-token
      annotations:
        kubernetes.io/service-account.name: vector-events
    type: kubernetes.io/service-account-token