scraper.go

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
package vcenterreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/vcenterreceiver"
import (
	"context"
	"fmt"

	"github.com/vmware/govmomi/performance"
	"github.com/vmware/govmomi/vim25/mo"
	"github.com/vmware/govmomi/vim25/types"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/receiver"
	"go.opentelemetry.io/collector/receiver/scrapererror"
	"go.uber.org/zap"

	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/vcenterreceiver/internal/metadata"
)

var _ receiver.Metrics = (*vcenterMetricScraper)(nil)

type vmGroupInfo struct {
	poweredOn  int64
	poweredOff int64
	suspended  int64
	templates  int64
}

type vcenterScrapeData struct {
	datacenters          []*mo.Datacenter
	datastores           []*mo.Datastore
	rPoolIPathsByRef     map[string]*string
	vAppIPathsByRef      map[string]*string
	rPoolsByRef          map[string]*mo.ResourcePool
	computesByRef        map[string]*mo.ComputeResource
	hostsByRef           map[string]*mo.HostSystem
	hostPerfMetricsByRef map[string]*performance.EntityMetric
	vmsByRef             map[string]*mo.VirtualMachine
	vmPerfMetricsByRef   map[string]*performance.EntityMetric
}

type vcenterMetricScraper struct {
	client     *vcenterClient
	config     *Config
	mb         *metadata.MetricsBuilder
	logger     *zap.Logger
	scrapeData *vcenterScrapeData
}

func newVmwareVcenterScraper(
	logger *zap.Logger,
	config *Config,
	settings receiver.CreateSettings,
) *vcenterMetricScraper {
	client := newVcenterClient(config)
	scrapeData := newVcenterScrapeData()

	return &vcenterMetricScraper{
		client:     client,
		config:     config,
		logger:     logger,
		mb:         metadata.NewMetricsBuilder(config.MetricsBuilderConfig, settings),
		scrapeData: scrapeData,
	}
}

func newVcenterScrapeData() *vcenterScrapeData {
	return &vcenterScrapeData{
		datacenters:          make([]*mo.Datacenter, 0),
		datastores:           make([]*mo.Datastore, 0),
		rPoolIPathsByRef:     make(map[string]*string),
		vAppIPathsByRef:      make(map[string]*string),
		computesByRef:        make(map[string]*mo.ComputeResource),
		hostsByRef:           make(map[string]*mo.HostSystem),
		hostPerfMetricsByRef: make(map[string]*performance.EntityMetric),
		rPoolsByRef:          make(map[string]*mo.ResourcePool),
		vmsByRef:             make(map[string]*mo.VirtualMachine),
		vmPerfMetricsByRef:   make(map[string]*performance.EntityMetric),
	}
}

func (v *vcenterMetricScraper) Start(ctx context.Context, _ component.Host) error {
	connectErr := v.client.EnsureConnection(ctx)
	// don't fail to start if we cannot establish connection, just log an error
	if connectErr != nil {
		v.logger.Error(fmt.Sprintf("unable to establish a connection to the vSphere SDK %s", connectErr.Error()))
	}
	return nil
}
func (v *vcenterMetricScraper) Shutdown(ctx context.Context) error {
	return v.client.Disconnect(ctx)
}
func (v *vcenterMetricScraper) scrape(ctx context.Context) (pmetric.Metrics, error) {
	if v.client == nil {
		v.client = newVcenterClient(v.config)
	}
	// ensure connection before scraping
	if err := v.client.EnsureConnection(ctx); err != nil {
		return pmetric.NewMetrics(), fmt.Errorf("unable to connect to vSphere SDK: %w", err)
	}

	errs := &scrapererror.ScrapeErrors{}
	err := v.scrapeAndProcessAllMetrics(ctx, errs)

	return v.mb.Emit(), err
}

// scrapeAndProcessAllMetrics collects & converts all relevant resources managed by vCenter to OTEL resources & metrics
func (v *vcenterMetricScraper) scrapeAndProcessAllMetrics(ctx context.Context, errs *scrapererror.ScrapeErrors) error {
	v.scrapeData = newVcenterScrapeData()
	v.scrapeResourcePoolInventoryListObjects(ctx, errs)
	v.scrapeVAppInventoryListObjects(ctx, errs)
	v.scrapeDatacenters(ctx, errs)

	for _, dc := range v.scrapeData.datacenters {
		v.scrapeDatastores(ctx, dc, errs)
		v.scrapeComputes(ctx, dc, errs)
		v.scrapeHosts(ctx, dc, errs)
		v.scrapeResourcePools(ctx, dc, errs)
		v.scrapeVirtualMachines(ctx, dc, errs)

		// Build metrics now that all vCenter data has been scraped for a single datacenter
		v.processDatacenterData(dc, errs)
	}
	// Clear scrape data
	v.scrapeData = nil

	return errs.Combine()
}

// scrapeResourcePoolInventoryListObjects scrapes and store all ResourcePool objects with their InventoryLists
func (v *vcenterMetricScraper) scrapeResourcePoolInventoryListObjects(ctx context.Context, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.rPoolIPathsByRef = make(map[string]*string)

	// Get ResourcePools with InventoryLists and store for later retrieval
	rPools, err := v.client.ResourcePoolInventoryListObjects(ctx)
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	for i := range rPools {
		v.scrapeData.rPoolIPathsByRef[rPools[i].Reference().Value] = &rPools[i].InventoryPath
	}
}

// scrapeVAppInventoryListObjects scrapes and stores all vApp objects with their InventoryLists
func (v *vcenterMetricScraper) scrapeVAppInventoryListObjects(ctx context.Context, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.vAppIPathsByRef = make(map[string]*string)

	// Get vApps with InventoryLists and store for later retrieval
	vApps, err := v.client.VAppInventoryListObjects(ctx)
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	for i := range vApps {
		v.scrapeData.vAppIPathsByRef[vApps[i].Reference().Value] = &vApps[i].InventoryPath
	}
}

// scrapeDatacenters scrapes and stores all relevant property data for all Datacenters
func (v *vcenterMetricScraper) scrapeDatacenters(ctx context.Context, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.datacenters = make([]*mo.Datacenter, 0)

	// Get Datacenters w/properties and store for later retrieval
	datacenters, err := v.client.Datacenters(ctx)
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	for i := range datacenters {
		v.scrapeData.datacenters = append(v.scrapeData.datacenters, &datacenters[i])
	}
}

// scrapeDatastores scrapes and stores all relevant property data for a Datacenter's Datastores
func (v *vcenterMetricScraper) scrapeDatastores(ctx context.Context, dc *mo.Datacenter, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.datastores = make([]*mo.Datastore, 0)

	// Get Datastores w/properties and store for later retrieval
	datastores, err := v.client.Datastores(ctx, dc.Reference())
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	for i := range datastores {
		v.scrapeData.datastores = append(v.scrapeData.datastores, &datastores[i])
	}
}

// scrapeComputes scrapes and stores all relevant property data for a Datacenter's ComputeResources/ClusterComputeResources
func (v *vcenterMetricScraper) scrapeComputes(ctx context.Context, dc *mo.Datacenter, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.computesByRef = make(map[string]*mo.ComputeResource)

	// Get ComputeResources/ClusterComputeResources w/properties and store for later retrieval
	computes, err := v.client.ComputeResources(ctx, dc.Reference())
	if err != nil {
		errs.AddPartial(1, err)
		return
	}

	for i := range computes {
		v.scrapeData.computesByRef[computes[i].Reference().Value] = &computes[i]
	}
}

// scrapeHosts scrapes and stores all relevant metric/property data for a Datacenter's HostSystems
func (v *vcenterMetricScraper) scrapeHosts(ctx context.Context, dc *mo.Datacenter, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.hostsByRef = make(map[string]*mo.HostSystem)

	// Get HostSystems w/properties and store for later retrieval
	hosts, err := v.client.HostSystems(ctx, dc.Reference())
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	hsRefs := []types.ManagedObjectReference{}
	for i := range hosts {
		hsRefs = append(hsRefs, hosts[i].Reference())
		v.scrapeData.hostsByRef[hosts[i].Reference().Value] = &hosts[i]
	}

	spec := types.PerfQuerySpec{
		MaxSample: 5,
		Format:    string(types.PerfFormatNormal),
		// Just grabbing real time performance metrics of the current
		// supported metrics by this receiver. If more are added we may need
		// a system of making this user customizable or adapt to use a 5 minute interval per metric
		IntervalId: int32(20),
	}
	// Get all HostSystem performance metrics and store for later retrieval
	results, err := v.client.PerfMetricsQuery(ctx, spec, hostPerfMetricList, hsRefs)
	if err != nil {
		errs.AddPartial(1, fmt.Errorf("failed to retrieve perf metrics for HostSystems: %w", err))
		return
	}
	v.scrapeData.hostPerfMetricsByRef = results.resultsByRef
}

// scrapeResourcePools scrapes and stores all relevant property data for a Datacenter's ResourcePools/vApps
func (v *vcenterMetricScraper) scrapeResourcePools(ctx context.Context, dc *mo.Datacenter, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.rPoolsByRef = make(map[string]*mo.ResourcePool)

	// Get ResourcePools/vApps w/properties and store for later retrieval
	rPools, err := v.client.ResourcePools(ctx, dc.Reference())
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	for i := range rPools {
		v.scrapeData.rPoolsByRef[rPools[i].Reference().Value] = &rPools[i]
	}
}

// scrapeVirtualMachines scrapes and stores all relevant metric/property data for a Datacenter's VirtualMachines
func (v *vcenterMetricScraper) scrapeVirtualMachines(ctx context.Context, dc *mo.Datacenter, errs *scrapererror.ScrapeErrors) {
	// Init for current collection
	v.scrapeData.vmsByRef = make(map[string]*mo.VirtualMachine)

	// Get VirtualMachines w/properties and store for later retrieval
	vms, err := v.client.VMs(ctx, dc.Reference())
	if err != nil {
		errs.AddPartial(1, err)
		return
	}
	vmRefs := []types.ManagedObjectReference{}
	for i := range vms {
		vmRefs = append(vmRefs, vms[i].Reference())
		v.scrapeData.vmsByRef[vms[i].Reference().Value] = &vms[i]
	}

	spec := types.PerfQuerySpec{
		Format: string(types.PerfFormatNormal),
		// Just grabbing real time performance metrics of the current
		// supported metrics by this receiver. If more are added we may need
		// a system of making this user customizable or adapt to use a 5 minute interval per metric
		IntervalId: int32(20),
	}
	// Get all VirtualMachine performance metrics and store for later retrieval
	results, err := v.client.PerfMetricsQuery(ctx, spec, vmPerfMetricList, vmRefs)
	if err != nil {
		errs.AddPartial(1, fmt.Errorf("failed to retrieve perf metrics for VirtualMachines: %w", err))
		return
	}
	v.scrapeData.vmPerfMetricsByRef = results.resultsByRef
}