Vcenterreceiver spike

receiver/vcenterreceiver/client.go

@@ -7,16 +7,172 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"time"
 
+	"github.com/ReneKroon/ttlcache/v2"
 	"github.com/vmware/govmomi"
 	"github.com/vmware/govmomi/find"
 	"github.com/vmware/govmomi/object"
 	"github.com/vmware/govmomi/performance"
 	"github.com/vmware/govmomi/property"
 	"github.com/vmware/govmomi/vim25"
 	vt "github.com/vmware/govmomi/vim25/types"
+	"go.uber.org/zap"
 )
 
+type client interface {
+	// EnsureConnection will establish a connection to the vSphere SDK if not already established
+	EnsureConnection(ctx context.Context, logger *zap.Logger) error
+	// Datacenters returns the datacenterComputeResources of the vSphere SDK
+	Datacenters(ctx context.Context) ([]*object.Datacenter, error)
+	// Clusters returns the clusterComputeResources of the vSphere SDK
+	Clusters(ctx context.Context, datacenter *object.Datacenter) ([]*object.ClusterComputeResource, error)
+	// ResourcePools returns the resourcePools in the vSphere SDK
+	ResourcePools(ctx context.Context) ([]*object.ResourcePool, error)
+	VMs(ctx context.Context) ([]*object.VirtualMachine, error)
+	Disconnect(ctx context.Context) error
+
+	PerformanceQuery(
+		ctx context.Context,
+		spec vt.PerfQuerySpec,
+		names []string,
+		objs []vt.ManagedObjectReference,
+	) (*perfSampleResult, error)
+}
+
+func newCachingClient(c *Config) client {
+	cache := ttlcache.NewCache()
+	_ = cache.SetTTL(c.CachingTTL)
+	cache.SkipTTLExtensionOnHit(true)
+	return &cachingClient{
+		delegate:   defaultNewVcenterClient(c),
+		cache:      cache,
+		refreshTTL: c.RefreshTTL,
+	}
+}
+
+type cachingClient struct {
+	delegate    client
+	cache       *ttlcache.Cache
+	refreshTTL  time.Duration
+	refreshChan chan struct{}
+}
+
+func (c *cachingClient) PerformanceQuery(ctx context.Context, spec vt.PerfQuerySpec, names []string, objs []vt.ManagedObjectReference) (*perfSampleResult, error) {
+	return c.delegate.PerformanceQuery(ctx, spec, names, objs)
+}
+
+func (c *cachingClient) EnsureConnection(ctx context.Context, logger *zap.Logger) error {
+	if vc, ok := c.delegate.(*vcenterClient); ok {
+		if vc.moClient != nil {
+			if sessionActive, _ := vc.moClient.SessionManager.SessionIsActive(ctx); sessionActive {
+				return nil
+			}
+		}
+	}
+	if c.refreshChan != nil {
+		close(c.refreshChan)
+	}
+	c.refreshChan = make(chan struct{})
+	if err := c.delegate.EnsureConnection(ctx, logger); err != nil {
+		return err
+	}
+	if err := c.cache.Purge(); err != nil {
+		return err
+	}
+	if err := c.refresh(ctx); err != nil {
+		return err
+	}
+	go func() {
+		ticker := time.NewTicker(c.refreshTTL)
+		for {
+			select {
+			case <-c.refreshChan:
+				return
+			case <-ticker.C:
+				err := c.refresh(ctx)
+				if err != nil {
+					logger.Error("Error refreshing vCenter data: %v")
+				}
+			}
+		}
+	}()
+	return nil
+}
+
+func (c *cachingClient) refresh(ctx context.Context) error {
+	// cache data centers
+	dcs, err := c.delegate.Datacenters(ctx)
+	if err != nil {
+		return err
+	}
+	if err = c.cache.Set("dcs", dcs); err != nil {
+		return err
+	}
+	// cache resource pools
+	rps, err := c.delegate.ResourcePools(ctx)
+	if err != nil {
+		return err
+	}
+	if err = c.cache.Set("resourcepools", rps); err != nil {
+		return err
+	}
+	// cache VMs
+	vms, err := c.delegate.VMs(ctx)
+	if err != nil {
+		return err
+	}
+	if err = c.cache.Set("vms", vms); err != nil {
+		return err
+	}
+
+	// cache clusters under data centers
+	for _, datacenter := range dcs {
+		key := "datacenter" + datacenter.Reference().String()
+
+		clusters, err := c.delegate.Clusters(ctx, datacenter)
+		if err != nil {
+			return err
+		}
+		if err = c.cache.Set(key, clusters); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *cachingClient) Datacenters(_ context.Context) ([]*object.Datacenter, error) {
+	dcs, _ := c.cache.Get("dcs")
+	return dcs.([]*object.Datacenter), nil
+}
+
+func (c *cachingClient) Clusters(_ context.Context, datacenter *object.Datacenter) ([]*object.ClusterComputeResource, error) {
+	key := "datacenter" + datacenter.Reference().String()
+	clusters, _ := c.cache.Get(key)
+	return clusters.([]*object.ClusterComputeResource), nil
+}
+
+func (c *cachingClient) ResourcePools(_ context.Context) ([]*object.ResourcePool, error) {
+	rps, _ := c.cache.Get("resourcepools")
+	return rps.([]*object.ResourcePool), nil
+}
+
+func (c *cachingClient) VMs(_ context.Context) ([]*object.VirtualMachine, error) {
+	vms, _ := c.cache.Get("vms")
+	return vms.([]*object.VirtualMachine), nil
+}
+
+func (c *cachingClient) Disconnect(ctx context.Context) error {
+	if c.refreshChan != nil {
+		close(c.refreshChan)
+	}
+	_ = c.cache.Close()
+	return c.delegate.Disconnect(ctx)
+}
+
+var _ client = &cachingClient{}
+
 // vcenterClient is a client that collects data from a vCenter endpoint.
 type vcenterClient struct {
 	moClient  *govmomi.Client
@@ -27,16 +183,16 @@ type vcenterClient struct {
 	cfg       *Config
 }
 
-var newVcenterClient = defaultNewVcenterClient
+var newVcenterClient = newCachingClient
 
-func defaultNewVcenterClient(c *Config) *vcenterClient {
+func defaultNewVcenterClient(c *Config) client {
 	return &vcenterClient{
 		cfg: c,
 	}
 }
 
 // EnsureConnection will establish a connection to the vSphere SDK if not already established
-func (vc *vcenterClient) EnsureConnection(ctx context.Context) error {
+func (vc *vcenterClient) EnsureConnection(ctx context.Context, _ *zap.Logger) error {
 	if vc.moClient != nil {
 		sessionActive, _ := vc.moClient.SessionManager.SessionIsActive(ctx)
 		if sessionActive {
@@ -121,7 +277,7 @@ type perfSampleResult struct {
 	results  []performance.EntityMetric
 }
 
-func (vc *vcenterClient) performanceQuery(
+func (vc *vcenterClient) PerformanceQuery(
 	ctx context.Context,
 	spec vt.PerfQuerySpec,
 	names []string,

receiver/vcenterreceiver/client_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/vmware/govmomi/vim25"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
+	"go.uber.org/zap"
 )
 
 func TestGetClusters(t *testing.T) {
@@ -86,7 +87,7 @@ func TestSessionReestablish(t *testing.T) {
 		require.NoError(t, err)
 		require.False(t, connected)
 
-		err = client.EnsureConnection(ctx)
+		err = client.EnsureConnection(ctx, zap.NewNop())
 		require.NoError(t, err)
 
 		connected, err = client.moClient.SessionManager.SessionIsActive(ctx)

receiver/vcenterreceiver/config.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"time"
 
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
@@ -24,6 +25,8 @@ type Config struct {
 	Endpoint                                string              `mapstructure:"endpoint"`
 	Username                                string              `mapstructure:"username"`
 	Password                                configopaque.String `mapstructure:"password"`
+	RefreshTTL                              time.Duration       `mapstructure:"refresh_ttl"`
+	CachingTTL                              time.Duration       `mapstructure:"caching_ttl"`
 }
 
 // Validate checks to see if the supplied config will work for the receiver

receiver/vcenterreceiver/factory.go

@@ -34,6 +34,8 @@ func createDefaultConfig() component.Config {
 		ScraperControllerSettings: cfg,
 		TLSClientSetting:          configtls.TLSClientSetting{},
 		MetricsBuilderConfig:      metadata.DefaultMetricsBuilderConfig(),
+		CachingTTL:                15 * time.Minute,
+		RefreshTTL:                5 * time.Minute,
 	}
 }
 

receiver/vcenterreceiver/go.mod

@@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/vcente
 go 1.20
 
 require (
+	github.com/ReneKroon/ttlcache/v2 v2.11.0
 	github.com/basgys/goxml2json v1.1.0
 	github.com/google/go-cmp v0.6.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.92.0
@@ -99,6 +100,7 @@ require (
 	golang.org/x/exp v0.0.0-20230711023510-fffb14384f22 // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/net v0.19.0 // indirect
+	golang.org/x/sync v0.5.0 // indirect
 	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.15.0 // indirect