func (c *Updater) handleCheckUpdateWithLock(ctx context.Context, check sm.Check) error {
	scraper, found := c.scrapers[check.Id]
	if !found {
		c.logger.Warn().Int64("check_id", check.Id).Msg("update request for an unknown check")
		return c.addAndStartScraperWithLock(ctx, check)
	}
	scraper.Stop()
	checkType := scraper.CheckType().String()
	delete(c.scrapers, check.Id)
	c.metrics.runningScrapers.WithLabelValues(checkType).Dec()
	return c.addAndStartScraperWithLock(ctx, check)
}
handleCheckUpdateWithLock is the bottom half of handleCheckUpdate. It
MUST be called with the scrapersMutex lock held.
package checks
import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
	"github.com/grafana/synthetic-monitoring-agent/internal/feature"
	"github.com/grafana/synthetic-monitoring-agent/internal/pkg/logproto"
	"github.com/grafana/synthetic-monitoring-agent/internal/pusher"
	"github.com/grafana/synthetic-monitoring-agent/internal/scraper"
	"github.com/grafana/synthetic-monitoring-agent/internal/version"
	sm "github.com/grafana/synthetic-monitoring-agent/pkg/pb/synthetic_monitoring"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/prometheus/prompb"
	"github.com/rs/zerolog"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)
var (
	errNotAuthorized     = errors.New("probe not authorized")
	errTransportClosing  = errors.New("transport closing")
	errProbeUnregistered = errors.New("probe no longer registered")
)
type Updater struct {
	api            apiInfo
	logger         zerolog.Logger
	features       feature.Collection
	publishCh      chan<- pusher.Payload
	tenantCh       chan<- sm.Tenant
	probe          *sm.Probe
	scrapersMutex  sync.Mutex
	scrapers       map[int64]*scraper.Scraper
	metrics        metrics
	scraperFactory func(context.Context, sm.Check, chan<- pusher.Payload, sm.Probe, zerolog.Logger, prometheus.Counter, *prometheus.CounterVec) (*scraper.Scraper, error)
}
type apiInfo struct {
	conn *grpc.ClientConn
}
type metrics struct {
	changeErrorsCounter *prometheus.CounterVec
	changesCounter      *prometheus.CounterVec
	connectionStatus    prometheus.Gauge
	probeInfo           *prometheus.GaugeVec
	runningScrapers     *prometheus.GaugeVec
	scrapeErrorCounter  *prometheus.CounterVec
	scrapesCounter      *prometheus.CounterVec
}
type TimeSeries = []prompb.TimeSeries
type Streams = []logproto.Stream
type UpdaterOptions struct {
	Conn           *grpc.ClientConn
	Logger         zerolog.Logger
	PublishCh      chan<- pusher.Payload
	TenantCh       chan<- sm.Tenant
	PromRegisterer prometheus.Registerer
	Features       feature.Collection
	ScraperFactory func(context.Context, sm.Check, chan<- pusher.Payload, sm.Probe, zerolog.Logger, prometheus.Counter, *prometheus.CounterVec) (*scraper.Scraper, error)
}
func NewUpdater(opts UpdaterOptions) (*Updater, error) {
	changesCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "sm_agent",
		Subsystem: "updater",
		Name:      "changes_total",
		Help:      "Total number of changes processed.",
	}, []string{
		"type",
	})
	if err := opts.PromRegisterer.Register(changesCounter); err != nil {
		return nil, err
	}
	changeErrorsCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "sm_agent",
		Subsystem: "updater",
		Name:      "change_errors_total",
		Help:      "Total number of errors during change processing.",
	}, []string{
		"type",
	})
	if err := opts.PromRegisterer.Register(changeErrorsCounter); err != nil {
		return nil, err
	}
	runningScrapers := prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "sm_agent",
		Subsystem: "updater",
		Name:      "scrapers_total",
		Help:      "Total number of running scrapers.",
	}, []string{
		"type",
	})
	if err := opts.PromRegisterer.Register(runningScrapers); err != nil {
		return nil, err
	}
	scrapesCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "sm_agent",
		Subsystem: "scraper",
		Name:      "operations_total",
		Help:      "Total number of scrape operations performed by type.",
	}, []string{
		"type",
	})
	if err := opts.PromRegisterer.Register(scrapesCounter); err != nil {
		return nil, err
	}
	scrapeErrorCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "sm_agent",
		Subsystem: "scraper",
		Name:      "errors_total",
		Help:      "Total number of scraper errors by type and status.",
	}, []string{
		"type",
		"source",
	})
	if err := opts.PromRegisterer.Register(scrapeErrorCounter); err != nil {
		return nil, err
	}
	connectionStatusGauge := prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "sm_agent",
		Subsystem: "api_connection",
		Name:      "status",
		Help:      "API connection status.",
	})
	if err := opts.PromRegisterer.Register(connectionStatusGauge); err != nil {
		return nil, err
	}
	connectionStatusGauge.Set(0)
	probeInfoGauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "sm_agent",
		Name:      "info",
		Help:      "Agent information.",
	}, []string{
		"id",
		"name",
		"version",
		"commit",
		"buildstamp",
	})
	if err := opts.PromRegisterer.Register(probeInfoGauge); err != nil {
		return nil, err
	}
	scraperFactory := scraper.New
	if opts.ScraperFactory != nil {
		scraperFactory = opts.ScraperFactory
	}
	return &Updater{
		api: apiInfo{
			conn: opts.Conn,
		},
		logger:         opts.Logger,
		features:       opts.Features,
		publishCh:      opts.PublishCh,
		tenantCh:       opts.TenantCh,
		scrapers:       make(map[int64]*scraper.Scraper),
		scraperFactory: scraperFactory,
		metrics: metrics{
			changeErrorsCounter: changeErrorsCounter,
			changesCounter:      changesCounter,
			connectionStatus:    connectionStatusGauge,
			probeInfo:           probeInfoGauge,
			runningScrapers:     runningScrapers,
			scrapeErrorCounter:  scrapeErrorCounter,
			scrapesCounter:      scrapesCounter,
		},
	}, nil
}
func (c *Updater) Run(ctx context.Context) error {
	for {
		err := c.loop(ctx)
		switch {
		case err == nil:
			return nil
		case errors.Is(err, errNotAuthorized):
			c.logger.Error().
				Err(err).
				Str("connection_state", c.api.conn.GetState().String()).
				Msg("cannot connect, bailing out")
			return err
		case errors.Is(err, errTransportClosing):
			c.logger.Warn().
				Err(err).
				Str("connection_state", c.api.conn.GetState().String()).
				Msg("the other end closed the connection, trying to reconnect")
			continue
		case errors.Is(err, context.Canceled):
			c.logger.Error().
				Err(err).
				Str("connection_state", c.api.conn.GetState().String()).
				Msg("context cancelled, closing updater")
			return nil
		case errors.Is(err, errProbeUnregistered):
			c.logger.Warn().
				Str("connection_state", c.api.conn.GetState().String()).
				Msg("unregistered probe in API, sleeping for 1 minute...")
			if err := sleepCtx(ctx, 1*time.Minute); err != nil {
				return err
			}
		default:
			c.logger.Warn().
				Err(err).
				Str("connection_state", c.api.conn.GetState().String()).
				Msg("handling check changes")
			if err := sleepCtx(ctx, 2*time.Second); err != nil {
				return err
			}
		}
	}
}
func (c *Updater) loop(ctx context.Context) error {
	c.logger.Info().Msg("fetching check configuration from synthetic-monitoring-api")
	client := sm.NewChecksClient(c.api.conn)
	grpcErrorHandler := func(action string, err error) error {
		status, ok := status.FromError(err)
		c.logger.Error().Err(err).Str("action", action).Uint32("code", uint32(status.Code())).Msg(status.Message())
		switch {
		case !ok:
			return fmt.Errorf("%s: %w", action, err)
		case status.Code() == codes.Canceled:
			return context.Canceled
		case status.Message() == "transport is closing":
			return errTransportClosing
		case status.Code() == codes.PermissionDenied:
			return errNotAuthorized
		default:
			return status.Err()
		}
	}
	result, err := client.RegisterProbe(ctx, &sm.ProbeInfo{Version: version.Short(), Commit: version.Commit(), Buildstamp: version.Buildstamp()})
	if err != nil {
		return grpcErrorHandler("registering probe with synthetic-monitoring-api", err)
	}
	switch result.Status.Code {
	case sm.StatusCode_OK:
	case sm.StatusCode_NOT_AUTHORIZED:
		return errNotAuthorized
	default:
		return fmt.Errorf("registering probe with synthetic-monitoring-api, response: %s", result.Status.Message)
	}
	c.probe = &result.Probe
	c.logger.Info().Int64("probe id", c.probe.Id).Str("probe name", c.probe.Name).Msg("registered probe with synthetic-monitoring-api")
	c.metrics.connectionStatus.Set(1)
	defer c.metrics.connectionStatus.Set(0)
	c.metrics.probeInfo.Reset()
	c.metrics.probeInfo.With(map[string]string{
		"id":         strconv.FormatInt(c.probe.Id, 10),
		"name":       c.probe.Name,
		"version":    version.Short(),
		"commit":     version.Commit(),
		"buildstamp": version.Buildstamp(),
	}).Set(1)
	sigCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	sigCtx, signalFired := installSignalHandler(sigCtx)
	action := "requesting changes from synthetic-monitoring-api"
	cc, err := client.GetChanges(sigCtx, &sm.Void{})
	if err == nil {
		action = "getting changes from synthetic-monitoring-api"
		err = c.processChanges(ctx, cc)
	}
	if err != nil {
		if atomic.LoadInt32(signalFired) == 1 {
			return errProbeUnregistered
		}
		return grpcErrorHandler(action, err)
	}
	return nil
}
func installSignalHandler(ctx context.Context) (context.Context, *int32) {
	sigCtx, cancel := context.WithCancel(ctx)
	fired := new(int32)
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGUSR1)
	go func() {
		select {
		case <-sigCh:
			atomic.StoreInt32(fired, 1)
			cancel()
		case <-ctx.Done():
		}
		signal.Stop(sigCh)
	}()
	return sigCtx, fired
}
func (c *Updater) processChanges(ctx context.Context, cc sm.Checks_GetChangesClient) error {
	firstBatch := true
	for {
		select {
		case <-cc.Context().Done():
			return nil
		case <-ctx.Done():
			return ctx.Err()
		default:
			switch msg, err := cc.Recv(); err {
			case nil:
				c.handleChangeBatch(ctx, msg, firstBatch)
				firstBatch = false
			case io.EOF:
				c.logger.Warn().Err(err).Msg("no more messages?")
				return nil
			default:
				return err
			}
		}
	}
}
func (c *Updater) handleCheckAdd(ctx context.Context, check sm.Check) error {
	c.metrics.changesCounter.WithLabelValues("add").Inc()
	if err := check.Validate(); err != nil {
		return fmt.Errorf("invalid check: %w", err)
	}
	c.scrapersMutex.Lock()
	defer c.scrapersMutex.Unlock()
	if running, found := c.scrapers[check.Id]; found {
		return fmt.Errorf("check with id %d already exists (version %s)", check.Id, running.ConfigVersion())
	}
	return c.addAndStartScraperWithLock(ctx, check)
}
func (c *Updater) handleCheckUpdate(ctx context.Context, check sm.Check) error {
	c.metrics.changesCounter.WithLabelValues("update").Inc()
	if err := check.Validate(); err != nil {
		return fmt.Errorf("invalid check: %w", err)
	}
	c.scrapersMutex.Lock()
	defer c.scrapersMutex.Unlock()
	return c.handleCheckUpdateWithLock(ctx, check)
}