Add a persistent cache for cloudfoundry metadata based on badger

CHANGELOG.next.asciidoc

@@ -448,6 +448,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Added experimental dataset `juniper/netscreen`. {pull}20820[20820]
 - Added experimental dataset `sophos/utm`. {pull}20820[20820]
 - Add Cloud Foundry tags in related events. {pull}21177[21177]
+- Cloud Foundry metadata is cached to disk. {pull}20775[20775]
 - Add option to select the type of index template to load: legacy, component, index. {pull}21212[21212]
 
 *Auditbeat*

go.mod

@@ -45,6 +45,7 @@ require (
  github.com/davecgh/go-xdr v0.0.0-20161123171359-e6a2ba005892 // indirect
  github.com/denisenkom/go-mssqldb v0.0.0-20200206145737-bbfc9a55622e
  github.com/devigned/tab v0.1.2-0.20190607222403-0c15cf42f9a2 // indirect
+ github.com/dgraph-io/badger/v2 v2.2007.2
  github.com/dgrijalva/jwt-go v3.2.1-0.20190620180102-5e25c22bd5d6+incompatible // indirect
  github.com/digitalocean/go-libvirt v0.0.0-20180301200012-6075ea3c39a1
  github.com/dlclark/regexp2 v1.1.7-0.20171009020623-7632a260cbaf // indirect
@@ -55,7 +56,7 @@ require (
  github.com/docker/go-units v0.4.0
  github.com/dop251/goja v0.0.0-20200831102558-9af81ddcf0e1
  github.com/dop251/goja_nodejs v0.0.0-20171011081505-adff31b136e6
- github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4
+ github.com/dustin/go-humanize v1.0.0
  github.com/eclipse/paho.mqtt.golang v1.2.1-0.20200121105743-0d940dd29fd2
  github.com/elastic/ecs v1.6.0
  github.com/elastic/elastic-agent-client/v7 v7.0.0-20200709172729-d43b7ad5833a
@@ -142,12 +143,13 @@ require (
  github.com/satori/go.uuid v1.2.0 // indirect
  github.com/shirou/gopsutil v2.19.11+incompatible
  github.com/shopspring/decimal v1.2.0
- github.com/spf13/cobra v0.0.3
+ github.com/spf13/cobra v0.0.5
  github.com/spf13/pflag v1.0.5
  github.com/stretchr/objx v0.2.0 // indirect
  github.com/stretchr/testify v1.6.1
  github.com/tsg/go-daemon v0.0.0-20200207173439-e704b93fd89b
  github.com/tsg/gopacket v0.0.0-20200626092518-2ab8e397a786
+ github.com/ugorji/go/codec v1.1.8
  github.com/urso/sderr v0.0.0-20200210124243-c2a16f3d43ec
  github.com/vmware/govmomi v0.0.0-20170802214208-2cad15190b41
  github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c

libbeat/tests/system/test_cmd_completion.py

@@ -17,7 +17,7 @@ def test_bash_completion(self):
  def test_zsh_completion(self):
  exit_code = self.run_beat(extra_args=["completion", "zsh"])
  assert exit_code == 0
- assert self.log_contains("#compdef mockbeat")
+ assert self.log_contains("#compdef _mockbeat mockbeat")
 
  def test_unknown_completion(self):
  exit_code = self.run_beat(extra_args=["completion", "awesomeshell"])

x-pack/libbeat/common/cloudfoundry/cache.go

@@ -5,13 +5,16 @@
 package cloudfoundry
 
 import (
+ "crypto/sha1"
+ "encoding/base64"
  "fmt"
  "time"
 
  "github.com/cloudfoundry-community/go-cfclient"
+ "github.com/pkg/errors"
 
- "github.com/elastic/beats/v7/libbeat/common"
  "github.com/elastic/beats/v7/libbeat/logp"
+ "github.com/elastic/beats/v7/x-pack/libbeat/persistentcache"
 )
 
 // cfClient interface is provided so unit tests can mock the actual client.
@@ -22,65 +25,113 @@ type cfClient interface {
 
 // clientCacheWrap wraps the cloudfoundry client to add a cache in front of GetAppByGuid.
 type clientCacheWrap struct {
- cache *common.Cache
+ cache *persistentcache.PersistentCache
  client cfClient
  log *logp.Logger
  errorTTL time.Duration
 }
 
 // newClientCacheWrap creates a new cache for application data.
-func newClientCacheWrap(client cfClient, ttl time.Duration, errorTTL time.Duration, log *logp.Logger) *clientCacheWrap {
+func newClientCacheWrap(client cfClient, cacheName string, ttl time.Duration, errorTTL time.Duration, log *logp.Logger) (*clientCacheWrap, error) {
+ options := persistentcache.Options{
+ Timeout: ttl,
+ }
+
+ name := "cloudfoundry"
+ if cacheName != "" {
+ name = name + "-" + sanitizeCacheName(cacheName)
+ }
+
+ cache, err := persistentcache.New(name, options)
+ if err != nil {
+ return nil, fmt.Errorf("creating metadata cache: %w", err)
+ }
+
  return &clientCacheWrap{
- cache: common.NewCacheWithExpireOnAdd(ttl, 100),
+ cache: cache,
  client: client,
  errorTTL: errorTTL,
  log: log,
- }
+ }, nil
 }
 
 type appResponse struct {
- app *cfclient.App
- err error
+ App AppMeta `json:"a"`
+ Error cfclient.CloudFoundryError `json:"e,omitempty"`
+ ErrorMessage string `json:"em,omitempty"`
+}
+
+func (r *appResponse) fromStructs(app cfclient.App, err error) {
+ if err != nil {
+ cause := errors.Cause(err)
+ if cferr, ok := cause.(cfclient.CloudFoundryError); ok {
+ r.Error = cferr
+ }
+ r.ErrorMessage = err.Error()
+ return
+ }
+ r.App = AppMeta{
+ Name: app.Name,
+ Guid: app.Guid,
+ SpaceName: app.SpaceData.Entity.Name,
+ SpaceGuid: app.SpaceData.Meta.Guid,
+ OrgName: app.SpaceData.Entity.OrgData.Entity.Name,
+ OrgGuid: app.SpaceData.Entity.OrgData.Meta.Guid,
+ }
+}
+
+func (r *appResponse) toStructs() (*AppMeta, error) {
+ var empty cfclient.CloudFoundryError
+ if r.Error != empty {
+ // Wrapping the error so cfclient.IsAppNotFoundError can identify it
+ return nil, errors.Wrap(r.Error, r.ErrorMessage)
+ }
+ if len(r.ErrorMessage) > 0 {
+ return nil, errors.New(r.ErrorMessage)
+ }
+ return &r.App, nil
 }
 
 // fetchApp uses the cfClient to retrieve an App entity and
 // stores it in the internal cache
-func (c *clientCacheWrap) fetchAppByGuid(guid string) (*cfclient.App, error) {
+func (c *clientCacheWrap) fetchAppByGuid(guid string) (*AppMeta, error) {
  app, err := c.client.GetAppByGuid(guid)
- resp := appResponse{
- app: &app,
- err: err,
- }
+ var resp appResponse
+ resp.fromStructs(app, err)
  timeout := time.Duration(0)
  if err != nil {
  // Cache nil, because is what we want to return when there was an error
- resp.app = nil
  timeout = c.errorTTL
  }
- c.cache.PutWithTimeout(guid, &resp, timeout)
- return resp.app, resp.err
+ err = c.cache.PutWithTimeout(guid, resp, timeout)
+ if err != nil {
+ return nil, fmt.Errorf("storing app response in cache: %w", err)
+ }
+ return resp.toStructs()
 }
 
 // GetApp returns CF Application info, either from the cache or
 // using the CF client.
-func (c *clientCacheWrap) GetAppByGuid(guid string) (*cfclient.App, error) {
- cachedResp := c.cache.Get(guid)
- if cachedResp == nil {
+func (c *clientCacheWrap) GetAppByGuid(guid string) (*AppMeta, error) {
+ var resp appResponse
+ err := c.cache.Get(guid, &resp)
+ if err != nil {
  return c.fetchAppByGuid(guid)
  }
- resp, ok := cachedResp.(*appResponse)
- if !ok {
- return nil, fmt.Errorf("error converting cached app response (of type %T), this is likely a bug", cachedResp)
- }
- return resp.app, resp.err
+ return resp.toStructs()
 }
 
-// StartJanitor starts a goroutine that will periodically clean the applications cache.
-func (c *clientCacheWrap) StartJanitor(interval time.Duration) {
- c.cache.StartJanitor(interval)
+// Close release resources associated with this client
+func (c *clientCacheWrap) Close() error {
+ err := c.cache.Close()
+ if err != nil {
+ return fmt.Errorf("closing cache: %w", err)
+ }
+ return nil
 }
 
-// StopJanitor stops the goroutine that periodically clean the applications cache.
-func (c *clientCacheWrap) StopJanitor() {
- c.cache.StopJanitor()
+// sanitizeCacheName returns a unique string that can be used safely as part of a file name
+func sanitizeCacheName(name string) string {
+ hash := sha1.Sum([]byte(name))
+ return base64.RawURLEncoding.EncodeToString(hash[:])
 }

x-pack/libbeat/common/cloudfoundry/cache_integration_test.go

@@ -31,7 +31,7 @@ func TestGetApps(t *testing.T) {
 
  client, err := hub.Client()
  require.NoError(t, err)
- apps, err := client.(*clientCacheWrap).client.(*cfclient.Client).ListApps()
+ apps, err := client.ListApps()
  require.NoError(t, err)
 
  t.Logf("%d applications available", len(apps))
@@ -40,8 +40,9 @@ func TestGetApps(t *testing.T) {
  if len(apps) == 0 {
  t.Skip("no apps in account?")
  }
- client, err := hub.Client()
+ client, err := hub.ClientWithCache()
  require.NoError(t, err)
+ defer client.Close()
 
  guid := apps[0].Guid
  app, err := client.GetAppByGuid(guid)
@@ -50,13 +51,15 @@ func TestGetApps(t *testing.T) {
  })
 
  t.Run("handle error when application is not available", func(t *testing.T) {
- client, err := hub.Client()
+ client, err := hub.ClientWithCache()
  require.NoError(t, err)
+ defer client.Close()
 
  testNotExists := func(t *testing.T) {
  app, err := client.GetAppByGuid("notexists")
  assert.Nil(t, app)
- assert.True(t, cfclient.IsAppNotFoundError(err))
+ assert.Error(t, err)
+ assert.True(t, cfclient.IsAppNotFoundError(err), "Error found: %v", err)
  }
 
  var firstTimeDuration time.Duration

x-pack/libbeat/common/cloudfoundry/cache_test.go

@@ -13,19 +13,25 @@ import (
  "github.com/cloudfoundry-community/go-cfclient"
  "github.com/gofrs/uuid"
  "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
 
  "github.com/elastic/beats/v7/libbeat/logp"
 )
 
 func TestClientCacheWrap(t *testing.T) {
- ttl := 500 * time.Millisecond
+ if testing.Short() {
+ t.Skip("skipping in short mode")
+ }
+
+ ttl := 2 * time.Second
  guid := mustCreateFakeGuid()
  app := cfclient.App{
- Guid:  guid,
- Memory: 1, // use this field to track if from cache or from client
+ Guid: guid,
+ Name: "Foo", // use this field to track if from cache or from client
  }
  fakeClient := &fakeCFClient{app, 0}
- cache := newClientCacheWrap(fakeClient, ttl, ttl, logp.NewLogger("cloudfoundry"))
+ cache, err := newClientCacheWrap(fakeClient, "test", ttl, ttl, logp.NewLogger("cloudfoundry"))
+ require.NoError(t, err)
 
  missingAppGuid := mustCreateFakeGuid()
 
@@ -44,25 +50,28 @@ func TestClientCacheWrap(t *testing.T) {
  // fetched from client for the first time
  one, err = cache.GetAppByGuid(guid)
  assert.NoError(t, err)
- assert.Equal(t, app, *one)
+ assert.Equal(t, app.Guid, one.Guid)
+ assert.Equal(t, app.Name, one.Name)
  assert.Equal(t, 2, fakeClient.callCount)
 
  // updated app in fake client, new fetch should not have updated app
  updatedApp := cfclient.App{
- Guid:  guid,
- Memory: 2,
+ Guid: guid,
+ Name: "Bar",
  }
  fakeClient.app = updatedApp
  two, err := cache.GetAppByGuid(guid)
  assert.NoError(t, err)
- assert.Equal(t, app, *two)
+ assert.Equal(t, app.Guid, two.Guid)
+ assert.Equal(t, app.Name, two.Name)
  assert.Equal(t, 2, fakeClient.callCount)
 
  // wait the ttl, then it should have updated app
  time.Sleep(ttl)
  three, err := cache.GetAppByGuid(guid)
  assert.NoError(t, err)
- assert.Equal(t, updatedApp, *three)
+ assert.Equal(t, updatedApp.Guid, three.Guid)
+ assert.Equal(t, updatedApp.Name, three.Name)
  assert.Equal(t, 3, fakeClient.callCount)
 }