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
}
installSignalHandler installs a signal handler for SIGUSR1.

The returned context's Done channel is closed if the signal is
delivered. To make it simpler to determine if the signal was
delivered, a value of 1 is written to the location pointed to by the
returned int32 pointer.

If the provided context's Done channel is closed before the signal is
delivered, the signal handler is removed and the returned context's
Done channel is closed, too. It's the callers responsibility to
cancel the provided context if it's no longer interested in the
signal.
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
}