Cherry-pick #17360 to 7.x: Add regex based cid extractor to add_process_metadata

CHANGELOG.next.asciidoc

@@ -269,6 +269,7 @@ https:/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add Kerberos support to Kafka input and output. {pull}16781[16781]
 - Update supported versions of `redis` output. {pull}17198[17198]
 - Update documentation for system.process.memory fields to include clarification on Windows os's. {pull}17268[17268]
+- Add optional regex based cid extractor to `add_kubernetes_metadata` processor. {pull}17360[17360]
 - Add `replace` processor for replacing string values of fields. {pull}17342[17342]
 - Add `urldecode` processor to for decoding URL-encoded fields. {pull}17505[17505]
 - Add support for AWS IAM `role_arn` in credentials config. {pull}17658[17658] {issue}12464[12464]

libbeat/processors/add_process_metadata/add_process_metadata.go

@@ -126,9 +126,9 @@ func newProcessMetadataProcessorWithProvider(cfg *common.Config, provider proces
 
  cgroupsCache := common.NewCacheWithRemovalListener(config.CgroupCacheExpireTime, 100, evictionListener)
  cgroupsCache.StartJanitor(config.CgroupCacheExpireTime)
- p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, processCgroupPaths, cgroupsCache)
+ p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, config.CgroupRegex, processCgroupPaths, cgroupsCache)
  } else {
- p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, processCgroupPaths, nil)
+ p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, config.CgroupRegex, processCgroupPaths, nil)
  }
 
  }

libbeat/processors/add_process_metadata/add_process_metadata_test.go

@@ -485,6 +485,31 @@ func TestAddProcessMetadata(t *testing.T) {
  },
  },
  },
+ {
+ description: "container.id based on regex in config",
+ config: common.MapStr{
+ "match_pids": []string{"system.process.ppid"},
+ "include_fields": []string{"container.id"},
+ "cgroup_regex": "\\/.+\\/.+\\/.+\\/([0-9a-f]{64}).*",
+ },
+ event: common.MapStr{
+ "system": common.MapStr{
+ "process": common.MapStr{
+ "ppid": "1",
+ },
+ },
+ },
+ expected: common.MapStr{
+ "system": common.MapStr{
+ "process": common.MapStr{
+ "ppid": "1",
+ },
+ },
+ "container": common.MapStr{
+ "id": "b5285682fba7449c86452b89a800609440ecc88a7ba5f2d38bedfb85409b30b1",
+ },
+ },
+ },
  {
  description: "without cgroup cache",
  config: common.MapStr{

libbeat/processors/add_process_metadata/config.go

@@ -51,6 +51,9 @@ type config struct {
  // CgroupPrefix is the prefix where the container id is inside cgroup
  CgroupPrefixes []string `config:"cgroup_prefixes"`
 
+ // CgroupRegex is the regular expression that captures the container id from cgroup path
+ CgroupRegex string `config:"cgroup_regex"`
+
  // CgroupCacheExpireTime is the length of time before cgroup cache elements expire in seconds,
  // set to 0 to disable the cgroup cache
  CgroupCacheExpireTime time.Duration `config:"cgroup_cache_expire_time"`

libbeat/processors/add_process_metadata/docs/add_process_metadata.asciidoc

@@ -80,6 +80,11 @@ to `/kubepods` and `/docker`. This is the prefix where the container ID is
 inside cgroup. For different runtime configurations of Kubernetes or Docker,
 the `cgroup_prefixes` can be set to overwrite the defaults.
 
+`cgroup_regex`:: (Optional) By default, the container id is extracted from 
+cgroup file based on `cgroup_prefixes`. This can be overwritten by specifying 
+regular expression with capture group for capturing container id from cgroup 
+path. For example: `^\/.+\/.+\/.+\/([0-9a-f]{64}).*`
+
 `cgroup_cache_expire_time`:: (Optional) By default, the 
 `cgroup_cache_expire_time` is set to 30 seconds. This is the length of time
 before cgroup cache elements expire in seconds. It can be set to 0 to disable

libbeat/processors/add_process_metadata/gosigar_cid_provider.go

@@ -20,6 +20,7 @@ package add_process_metadata
 import (
  "os"
  "path/filepath"
+ "regexp"
  "strings"
 
  "github.com/pkg/errors"
@@ -36,6 +37,7 @@ type gosigarCidProvider struct {
  log *logp.Logger
  hostPath string
  cgroupPrefixes []string
+ cgroupRegex string
  processCgroupPaths func(string, int) (map[string]string, error)
  pidCidCache *common.Cache
 }
@@ -68,11 +70,12 @@ func (p gosigarCidProvider) GetCid(pid int) (result string, err error) {
  return cid, nil
 }
 
-func newCidProvider(hostPath string, cgroupPrefixes []string, processCgroupPaths func(string, int) (map[string]string, error), pidCidCache *common.Cache) gosigarCidProvider {
+func newCidProvider(hostPath string, cgroupPrefixes []string, cgroupRegex string, processCgroupPaths func(string, int) (map[string]string, error), pidCidCache *common.Cache) gosigarCidProvider {
  return gosigarCidProvider{
  log: logp.NewLogger(providerName),
  hostPath: hostPath,
  cgroupPrefixes: cgroupPrefixes,
+ cgroupRegex: cgroupRegex,
  processCgroupPaths: processCgroupPaths,
  pidCidCache: pidCidCache,
  }
@@ -104,12 +107,25 @@ func (p gosigarCidProvider) getProcessCgroups(pid int) (map[string]string, error
 // Example:
 // /kubepods/besteffort/pod9b9e44c2-00fd-11ea-95e9-080027421ddf/2bb9fd4de339e5d4f094e78bb87636004acfe53f5668104addc761fe4a93588e
 func (p gosigarCidProvider) getCid(cgroups map[string]string) string {
- for _, path := range cgroups {
- for _, prefix := range p.cgroupPrefixes {
- if strings.HasPrefix(path, prefix) {
- return filepath.Base(path)
+ // if regex defined use it to find cid
+ if len(p.cgroupRegex) != 0 {
+ re := regexp.MustCompile(p.cgroupRegex)
+ for _, path := range cgroups {
+ rs := re.FindStringSubmatch(path)
+ if rs != nil {
+ return rs[1]
  }
  }
+ } else {
+ // use string prefix to find cid
+ for _, path := range cgroups {
+ for _, prefix := range p.cgroupPrefixes {
+ if strings.HasPrefix(path, prefix) {
+ return filepath.Base(path)
+ }
+ }
+
+ }
  }
  return ""
 }