package telemetry

import (

"bytes"

"context"

"crypto/sha256"

"database/sql"

"encoding/json"

"errors"

"fmt"

"net"

"net/http"

"net/url"

"os"

"regexp"

"runtime"

"slices"

"strconv"

"strings"

"sync"

"time"

"github.com/elastic/go-sysinfo"

"github.com/google/uuid"

"golang.org/x/sync/errgroup"

"golang.org/x/xerrors"

"google.golang.org/protobuf/types/known/durationpb"

"google.golang.org/protobuf/types/known/wrapperspb"

"cdr.dev/slog/v3"

"github.com/coder/coder/v2/buildinfo"

clitelemetry "github.com/coder/coder/v2/cli/telemetry"

"github.com/coder/coder/v2/coderd/database"

"github.com/coder/coder/v2/coderd/database/dbauthz"

"github.com/coder/coder/v2/coderd/database/dbtime"

"github.com/coder/coder/v2/coderd/util/ptr"

"github.com/coder/coder/v2/codersdk"

tailnetproto "github.com/coder/coder/v2/tailnet/proto"

"github.com/coder/quartz"

)

const (

// VersionHeader is sent in every telemetry request to

// report the semantic version of Coder.

VersionHeader = "X-Coder-Version"

DefaultSnapshotFrequency = 30 * time.Minute

)

type Options struct {

Disabled bool

Database database.Store

Logger slog.Logger

Clock quartz.Clock

// URL is an endpoint to direct telemetry towards!

URL *url.URL

Experiments codersdk.Experiments

DeploymentID string

DeploymentConfig *codersdk.DeploymentValues

BuiltinPostgres bool

Tunnel bool

SnapshotFrequency time.Duration

ParseLicenseJWT func(lic *License) error

}

// New constructs a reporter for telemetry data.

// Duplicate data will be sent, it's on the server-side to index by UUID.

// Data is anonymized prior to being sent!

func New(options Options) (Reporter, error) {

if options.Clock == nil {

options.Clock = quartz.NewReal()

}

if options.SnapshotFrequency == 0 {

options.SnapshotFrequency = DefaultSnapshotFrequency

}

snapshotURL, err := options.URL.Parse("/snapshot")

if err != nil {

return nil, xerrors.Errorf("parse snapshot url: %w", err)

}

deploymentURL, err := options.URL.Parse("/deployment")

if err != nil {

return nil, xerrors.Errorf("parse deployment url: %w", err)

}

ctx, cancelFunc := context.WithCancel(context.Background())

reporter := &remoteReporter{

ctx: ctx,

closed: make(chan struct{}),

closeFunc: cancelFunc,

options: options,

deploymentURL: deploymentURL,

snapshotURL: snapshotURL,

startedAt: dbtime.Time(options.Clock.Now()).UTC(),

client: &http.Client{},

}

go reporter.runSnapshotter()

return reporter, nil

}

// NewNoop creates a new telemetry reporter that entirely discards all requests.

func NewNoop() Reporter {

return &noopReporter{}

}

// Reporter sends data to the telemetry server.

type Reporter interface {

// Report sends a snapshot to the telemetry server.

// The contents of the snapshot can be a partial representation of the

// database. For example, if a new user is added, a snapshot can

// contain just that user entry.

Report(snapshot *Snapshot)

Enabled() bool

Close()

}

type remoteReporter struct {

ctx context.Context

closed chan struct{}

closeMutex sync.Mutex

closeFunc context.CancelFunc

options Options

deploymentURL,

snapshotURL *url.URL

startedAt time.Time

shutdownAt *time.Time

client *http.Client

}

func (r *remoteReporter) Enabled() bool {

return !r.options.Disabled

}

func (r *remoteReporter) Report(snapshot *Snapshot) {

go r.reportSync(snapshot)

}

func (r *remoteReporter) reportSync(snapshot *Snapshot) {

snapshot.DeploymentID = r.options.DeploymentID

data, err := json.Marshal(snapshot)

if err != nil {

r.options.Logger.Error(r.ctx, "marshal snapshot: %w", slog.Error(err))

return

}

req, err := http.NewRequestWithContext(r.ctx, "POST", r.snapshotURL.String(), bytes.NewReader(data))

if err != nil {

r.options.Logger.Error(r.ctx, "unable to create snapshot request", slog.Error(err))

return

}

req.Header.Set(VersionHeader, buildinfo.Version())

resp, err := r.client.Do(req)

if err != nil {

// If the request fails it's not necessarily an error.

// In an airgapped environment, it's fine if this fails!

r.options.Logger.Debug(r.ctx, "submit", slog.Error(err))

return

}

defer resp.Body.Close()

if resp.StatusCode != http.StatusAccepted {

r.options.Logger.Debug(r.ctx, "bad response from telemetry server", slog.F("status", resp.StatusCode))

return

}

r.options.Logger.Debug(r.ctx, "submitted snapshot")

}

func (r *remoteReporter) Close() {

r.closeMutex.Lock()

defer r.closeMutex.Unlock()

if r.isClosed() {

return

}

close(r.closed)

now := dbtime.Time(r.options.Clock.Now()).UTC()

r.shutdownAt = &now

if r.Enabled() {

// Report a final collection of telemetry prior to close!

// This could indicate final actions a user has taken, and

// the time the deployment was shutdown.

r.reportWithDeployment()

}

r.closeFunc()

}

func (r *remoteReporter) isClosed() bool {

select {

case <-r.closed:

return true

default:

return false

}

}

// See the corresponding test in telemetry_test.go for a truth table.

func ShouldReportTelemetryDisabled(recordedTelemetryEnabled *bool, telemetryEnabled bool) bool {

return recordedTelemetryEnabled != nil && *recordedTelemetryEnabled && !telemetryEnabled

}

// RecordTelemetryStatus records the telemetry status in the database.

// If the status changed from enabled to disabled, returns a snapshot to

// be sent to the telemetry server.

func RecordTelemetryStatus( //nolint:revive

ctx context.Context,

logger slog.Logger,

db database.Store,

telemetryEnabled bool,

) (*Snapshot, error) {

item, err := db.GetTelemetryItem(ctx, string(TelemetryItemKeyTelemetryEnabled))

if err != nil && !errors.Is(err, sql.ErrNoRows) {

return nil, xerrors.Errorf("get telemetry enabled: %w", err)

}

var recordedTelemetryEnabled *bool

if !errors.Is(err, sql.ErrNoRows) {

value, err := strconv.ParseBool(item.Value)

if err != nil {

logger.Debug(ctx, "parse telemetry enabled", slog.Error(err))

}

// If ParseBool fails, value will default to false.

// This may happen if an admin manually edits the telemetry item

// in the database.

recordedTelemetryEnabled = &value

}

if err := db.UpsertTelemetryItem(ctx, database.UpsertTelemetryItemParams{

Key: string(TelemetryItemKeyTelemetryEnabled),

Value: strconv.FormatBool(telemetryEnabled),

}); err != nil {

return nil, xerrors.Errorf("upsert telemetry enabled: %w", err)

}

shouldReport := ShouldReportTelemetryDisabled(recordedTelemetryEnabled, telemetryEnabled)

if !shouldReport {

return nil, nil //nolint:nilnil

}

// If any of the following calls fail, we will never report that telemetry changed

// from enabled to disabled. This is okay. We only want to ping the telemetry server

// once, and never again. If that attempt fails, so be it.

item, err = db.GetTelemetryItem(ctx, string(TelemetryItemKeyTelemetryEnabled))

if err != nil {

return nil, xerrors.Errorf("get telemetry enabled after upsert: %w", err)

}

return &Snapshot{

TelemetryItems: []TelemetryItem{

ConvertTelemetryItem(item),

},

}, nil

}

func (r *remoteReporter) runSnapshotter() {

telemetryDisabledSnapshot, err := RecordTelemetryStatus(r.ctx, r.options.Logger, r.options.Database, r.Enabled())

if err != nil {

r.options.Logger.Debug(r.ctx, "record and maybe report telemetry status", slog.Error(err))

}

if telemetryDisabledSnapshot != nil {

r.reportSync(telemetryDisabledSnapshot)

}

r.options.Logger.Debug(r.ctx, "finished telemetry status check")

if !r.Enabled() {

return

}

first := true

ticker := time.NewTicker(r.options.SnapshotFrequency)

defer ticker.Stop()

for {

if !first {

select {

case <-r.closed:

return

case <-ticker.C:

}

// Skip the ticker on the first run to report instantly!

}

first = false

r.closeMutex.Lock()

if r.isClosed() {

r.closeMutex.Unlock()

return

}

r.reportWithDeployment()

r.closeMutex.Unlock()

}

}

func (r *remoteReporter) reportWithDeployment() {

// Submit deployment information before creating a snapshot!

// This is separated from the snapshot API call to reduce

// duplicate data from being inserted. Snapshot may be called

// numerous times simultaneously if there is lots of activity!

err := r.deployment()

if err != nil {

r.options.Logger.Debug(r.ctx, "update deployment", slog.Error(err))

return

}

snapshot, err := r.createSnapshot()

if errors.Is(err, context.Canceled) {

return

}

if err != nil {

r.options.Logger.Error(r.ctx, "unable to create deployment snapshot", slog.Error(err))

return

}

r.reportSync(snapshot)

}

// deployment collects host information and reports it to the telemetry server.

func (r *remoteReporter) deployment() error {

sysInfoHost, err := sysinfo.Host()

if err != nil {

return xerrors.Errorf("get host info: %w", err)

}

mem, err := sysInfoHost.Memory()

if err != nil {

return xerrors.Errorf("get memory info: %w", err)

}

sysInfo := sysInfoHost.Info()

containerized := false

if sysInfo.Containerized != nil {

containerized = *sysInfo.Containerized

}

// Tracks where Coder was installed from!

installSource := os.Getenv("CODER_TELEMETRY_INSTALL_SOURCE")

if len(installSource) > 64 {

return xerrors.Errorf("install source must be <=64 chars: %s", installSource)

}

idpOrgSync, err := checkIDPOrgSync(r.ctx, r.options.Database, r.options.DeploymentConfig)

if err != nil {

r.options.Logger.Debug(r.ctx, "check IDP org sync", slog.Error(err))

}

data, err := json.Marshal(&Deployment{

ID: r.options.DeploymentID,

Architecture: sysInfo.Architecture,

BuiltinPostgres: r.options.BuiltinPostgres,

Containerized: containerized,

Config: r.options.DeploymentConfig,

Kubernetes: os.Getenv("KUBERNETES_SERVICE_HOST") != "",

InstallSource: installSource,

Tunnel: r.options.Tunnel,

OSType: sysInfo.OS.Type,

OSFamily: sysInfo.OS.Family,

OSPlatform: sysInfo.OS.Platform,

OSName: sysInfo.OS.Name,

OSVersion: sysInfo.OS.Version,

CPUCores: runtime.NumCPU(),

MemoryTotal: mem.Total,

MachineID: sysInfo.UniqueID,

StartedAt: r.startedAt,

ShutdownAt: r.shutdownAt,

IDPOrgSync: &idpOrgSync,

})

if err != nil {

return xerrors.Errorf("marshal deployment: %w", err)

}

req, err := http.NewRequestWithContext(r.ctx, "POST", r.deploymentURL.String(), bytes.NewReader(data))

if err != nil {

return xerrors.Errorf("create deployment request: %w", err)

}

req.Header.Set(VersionHeader, buildinfo.Version())

resp, err := r.client.Do(req)

if err != nil {

return xerrors.Errorf("perform request: %w", err)

}

defer resp.Body.Close()

if resp.StatusCode != http.StatusAccepted {

return xerrors.Errorf("update deployment: %w", err)

}

r.options.Logger.Debug(r.ctx, "submitted deployment info")

return nil

}

// idpOrgSyncConfig is a subset of

// https://github.com/coder/coder/blob/5c6578d84e2940b9cfd04798c45e7c8042c3fe0e/coderd/idpsync/organization.go#L148

type idpOrgSyncConfig struct {

Field string `json:"field"`

}

// checkIDPOrgSync inspects the server flags and the runtime config. It's based on

// the OrganizationSyncEnabled function from enterprise/coderd/enidpsync/organizations.go.

// It has one distinct difference: it doesn't check if the license entitles to the

// feature, it only checks if the feature is configured.

//

// The above function is not used because it's very hard to make it available in

// the telemetry package due to coder/coder package structure and initialization

// order of the coder server.

//

// We don't check license entitlements because it's also hard to do from the

// telemetry package, and the config check should be sufficient for telemetry purposes.

//

// While this approach duplicates code, it's simpler than the alternative.

//

// See https://github.com/coder/coder/pull/16323 for more details.

func checkIDPOrgSync(ctx context.Context, db database.Store, values *codersdk.DeploymentValues) (bool, error) {

// key based on https://github.com/coder/coder/blob/5c6578d84e2940b9cfd04798c45e7c8042c3fe0e/coderd/idpsync/idpsync.go#L168

syncConfigRaw, err := db.GetRuntimeConfig(ctx, "organization-sync-settings")

if err != nil {

if errors.Is(err, sql.ErrNoRows) {

// If the runtime config is not set, we check if the deployment config

// has the organization field set.

return values != nil && values.OIDC.OrganizationField != "", nil

}

return false, xerrors.Errorf("get runtime config: %w", err)

}

syncConfig := idpOrgSyncConfig{}

if err := json.Unmarshal([]byte(syncConfigRaw), &syncConfig); err != nil {

return false, xerrors.Errorf("unmarshal runtime config: %w", err)

}

return syncConfig.Field != "", nil

}

// createSnapshot collects a full snapshot from the database.

func (r *remoteReporter) createSnapshot() (*Snapshot, error) {

var (

ctx = r.ctx

now = r.options.Clock.Now()

// For resources that grow in size very quickly (like workspace builds),

// we only report events that occurred within the past hour.

createdAfter = dbtime.Time(now.Add(-1 * time.Hour)).UTC()

eg errgroup.Group

snapshot = &Snapshot{

DeploymentID: r.options.DeploymentID,

}

)

eg.Go(func() error {

apiKeys, err := r.options.Database.GetAPIKeysLastUsedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get api keys last used: %w", err)

}

snapshot.APIKeys = make([]APIKey, 0, len(apiKeys))

for _, apiKey := range apiKeys {

snapshot.APIKeys = append(snapshot.APIKeys, ConvertAPIKey(apiKey))

}

return nil

})

eg.Go(func() error {

jobs, err := r.options.Database.GetProvisionerJobsCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get provisioner jobs: %w", err)

}

snapshot.ProvisionerJobs = make([]ProvisionerJob, 0, len(jobs))

for _, job := range jobs {

snapshot.ProvisionerJobs = append(snapshot.ProvisionerJobs, ConvertProvisionerJob(job))

}

return nil

})

eg.Go(func() error {

templates, err := r.options.Database.GetTemplates(ctx)

if err != nil {

return xerrors.Errorf("get templates: %w", err)

}

snapshot.Templates = make([]Template, 0, len(templates))

for _, dbTemplate := range templates {

snapshot.Templates = append(snapshot.Templates, ConvertTemplate(dbTemplate))

}

return nil

})

eg.Go(func() error {

templateVersions, err := r.options.Database.GetTemplateVersionsCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get template versions: %w", err)

}

snapshot.TemplateVersions = make([]TemplateVersion, 0, len(templateVersions))

for _, version := range templateVersions {

snapshot.TemplateVersions = append(snapshot.TemplateVersions, ConvertTemplateVersion(version))

}

return nil

})

eg.Go(func() error {

userRows, err := r.options.Database.GetUsers(ctx, database.GetUsersParams{})

if err != nil {

return xerrors.Errorf("get users: %w", err)

}

users := database.ConvertUserRows(userRows)

var firstUser database.User

for _, dbUser := range users {

if firstUser.CreatedAt.IsZero() {

firstUser = dbUser

}

if dbUser.CreatedAt.Before(firstUser.CreatedAt) {

firstUser = dbUser

}

}

snapshot.Users = make([]User, 0, len(users))

for _, dbUser := range users {

user := ConvertUser(dbUser)

// If it's the first user, we'll send the email!

if firstUser.ID == dbUser.ID {

email := dbUser.Email

user.Email = &email

}

snapshot.Users = append(snapshot.Users, user)

}

return nil

})

eg.Go(func() error {

groups, err := r.options.Database.GetGroups(ctx, database.GetGroupsParams{})

if err != nil {

return xerrors.Errorf("get groups: %w", err)

}

snapshot.Groups = make([]Group, 0, len(groups))

for _, group := range groups {

snapshot.Groups = append(snapshot.Groups, ConvertGroup(group.Group))

}

return nil

})

eg.Go(func() error {

groupMembers, err := r.options.Database.GetGroupMembers(ctx, false)

if err != nil {

return xerrors.Errorf("get groups: %w", err)

}

snapshot.GroupMembers = make([]GroupMember, 0, len(groupMembers))

for _, member := range groupMembers {

snapshot.GroupMembers = append(snapshot.GroupMembers, ConvertGroupMember(member))

}

return nil

})

eg.Go(func() error {

workspaceRows, err := r.options.Database.GetWorkspaces(ctx, database.GetWorkspacesParams{})

if err != nil {

return xerrors.Errorf("get workspaces: %w", err)

}

workspaces, err := database.ConvertWorkspaceRows(workspaceRows)

if err != nil {

return xerrors.Errorf("convert workspace rows: %w", err)

}

snapshot.Workspaces = make([]Workspace, 0, len(workspaces))

for _, dbWorkspace := range workspaces {

snapshot.Workspaces = append(snapshot.Workspaces, ConvertWorkspace(dbWorkspace))

}

return nil

})

eg.Go(func() error {

workspaceApps, err := r.options.Database.GetWorkspaceAppsCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace apps: %w", err)

}

snapshot.WorkspaceApps = make([]WorkspaceApp, 0, len(workspaceApps))

for _, app := range workspaceApps {

snapshot.WorkspaceApps = append(snapshot.WorkspaceApps, ConvertWorkspaceApp(app))

}

return nil

})

eg.Go(func() error {

workspaceAgents, err := r.options.Database.GetWorkspaceAgentsCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace agents: %w", err)

}

snapshot.WorkspaceAgents = make([]WorkspaceAgent, 0, len(workspaceAgents))

for _, agent := range workspaceAgents {

snapshot.WorkspaceAgents = append(snapshot.WorkspaceAgents, ConvertWorkspaceAgent(agent))

}

return nil

})

eg.Go(func() error {

workspaceBuilds, err := r.options.Database.GetWorkspaceBuildsCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace builds: %w", err)

}

snapshot.WorkspaceBuilds = make([]WorkspaceBuild, 0, len(workspaceBuilds))

for _, build := range workspaceBuilds {

snapshot.WorkspaceBuilds = append(snapshot.WorkspaceBuilds, ConvertWorkspaceBuild(build))

}

return nil

})

eg.Go(func() error {

workspaceResources, err := r.options.Database.GetWorkspaceResourcesCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace resources: %w", err)

}

snapshot.WorkspaceResources = make([]WorkspaceResource, 0, len(workspaceResources))

for _, resource := range workspaceResources {

snapshot.WorkspaceResources = append(snapshot.WorkspaceResources, ConvertWorkspaceResource(resource))

}

return nil

})

eg.Go(func() error {

workspaceMetadata, err := r.options.Database.GetWorkspaceResourceMetadataCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace resource metadata: %w", err)

}

snapshot.WorkspaceResourceMetadata = make([]WorkspaceResourceMetadata, 0, len(workspaceMetadata))

for _, metadata := range workspaceMetadata {

snapshot.WorkspaceResourceMetadata = append(snapshot.WorkspaceResourceMetadata, ConvertWorkspaceResourceMetadata(metadata))

}

return nil

})

eg.Go(func() error {

workspaceModules, err := r.options.Database.GetWorkspaceModulesCreatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace modules: %w", err)

}

snapshot.WorkspaceModules = make([]WorkspaceModule, 0, len(workspaceModules))

for _, module := range workspaceModules {

snapshot.WorkspaceModules = append(snapshot.WorkspaceModules, ConvertWorkspaceModule(module))

}

return nil

})

eg.Go(func() error {

licenses, err := r.options.Database.GetUnexpiredLicenses(ctx)

if err != nil {

return xerrors.Errorf("get licenses: %w", err)

}

snapshot.Licenses = make([]License, 0, len(licenses))

for _, license := range licenses {

tl := ConvertLicense(license)

if r.options.ParseLicenseJWT != nil {

if err := r.options.ParseLicenseJWT(&tl); err != nil {

r.options.Logger.Warn(ctx, "parse license JWT", slog.Error(err))

}

}

snapshot.Licenses = append(snapshot.Licenses, tl)

}

return nil

})

eg.Go(func() error {

if r.options.DeploymentConfig != nil && slices.Contains(r.options.DeploymentConfig.Experiments, string(codersdk.ExperimentWorkspaceUsage)) {

agentStats, err := r.options.Database.GetWorkspaceAgentUsageStats(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace agent stats: %w", err)

}

snapshot.WorkspaceAgentStats = make([]WorkspaceAgentStat, 0, len(agentStats))

for _, stat := range agentStats {

snapshot.WorkspaceAgentStats = append(snapshot.WorkspaceAgentStats, ConvertWorkspaceAgentStat(database.GetWorkspaceAgentStatsRow(stat)))

}

} else {

agentStats, err := r.options.Database.GetWorkspaceAgentStats(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get workspace agent stats: %w", err)

}

snapshot.WorkspaceAgentStats = make([]WorkspaceAgentStat, 0, len(agentStats))

for _, stat := range agentStats {

snapshot.WorkspaceAgentStats = append(snapshot.WorkspaceAgentStats, ConvertWorkspaceAgentStat(stat))

}

}

return nil

})

eg.Go(func() error {

memoryMonitors, err := r.options.Database.FetchMemoryResourceMonitorsUpdatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get memory resource monitors: %w", err)

}

snapshot.WorkspaceAgentMemoryResourceMonitors = make([]WorkspaceAgentMemoryResourceMonitor, 0, len(memoryMonitors))

for _, monitor := range memoryMonitors {

snapshot.WorkspaceAgentMemoryResourceMonitors = append(snapshot.WorkspaceAgentMemoryResourceMonitors, ConvertWorkspaceAgentMemoryResourceMonitor(monitor))

}

return nil

})

eg.Go(func() error {

volumeMonitors, err := r.options.Database.FetchVolumesResourceMonitorsUpdatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get volume resource monitors: %w", err)

}

snapshot.WorkspaceAgentVolumeResourceMonitors = make([]WorkspaceAgentVolumeResourceMonitor, 0, len(volumeMonitors))

for _, monitor := range volumeMonitors {

snapshot.WorkspaceAgentVolumeResourceMonitors = append(snapshot.WorkspaceAgentVolumeResourceMonitors, ConvertWorkspaceAgentVolumeResourceMonitor(monitor))

}

return nil

})

eg.Go(func() error {

proxies, err := r.options.Database.GetWorkspaceProxies(ctx)

if err != nil {

return xerrors.Errorf("get workspace proxies: %w", err)

}

snapshot.WorkspaceProxies = make([]WorkspaceProxy, 0, len(proxies))

for _, proxy := range proxies {

snapshot.WorkspaceProxies = append(snapshot.WorkspaceProxies, ConvertWorkspaceProxy(proxy))

}

return nil

})

eg.Go(func() error {

// Warning: When an organization is deleted, it's completely removed from

// the database. It will no longer be reported, and there will be no other

// indicator that it was deleted. This requires special handling when

// interpreting the telemetry data later.

orgs, err := r.options.Database.GetOrganizations(r.ctx, database.GetOrganizationsParams{})

if err != nil {

return xerrors.Errorf("get organizations: %w", err)

}

snapshot.Organizations = make([]Organization, 0, len(orgs))

for _, org := range orgs {

snapshot.Organizations = append(snapshot.Organizations, ConvertOrganization(org))

}

return nil

})

eg.Go(func() error {

items, err := r.options.Database.GetTelemetryItems(ctx)

if err != nil {

return xerrors.Errorf("get telemetry items: %w", err)

}

snapshot.TelemetryItems = make([]TelemetryItem, 0, len(items))

for _, item := range items {

snapshot.TelemetryItems = append(snapshot.TelemetryItems, ConvertTelemetryItem(item))

}

return nil

})

eg.Go(func() error {

metrics, err := r.options.Database.GetPrebuildMetrics(ctx)

if err != nil {

return xerrors.Errorf("get prebuild metrics: %w", err)

}

var totalCreated, totalFailed, totalClaimed int64

for _, metric := range metrics {

totalCreated += metric.CreatedCount

totalFailed += metric.FailedCount

totalClaimed += metric.ClaimedCount

}

snapshot.PrebuiltWorkspaces = make([]PrebuiltWorkspace, 0, 3)

now := dbtime.Now()

if totalCreated > 0 {

snapshot.PrebuiltWorkspaces = append(snapshot.PrebuiltWorkspaces, PrebuiltWorkspace{

ID: uuid.New(),

CreatedAt: now,

EventType: PrebuiltWorkspaceEventTypeCreated,

Count: int(totalCreated),

})

}

if totalFailed > 0 {

snapshot.PrebuiltWorkspaces = append(snapshot.PrebuiltWorkspaces, PrebuiltWorkspace{

ID: uuid.New(),

CreatedAt: now,

EventType: PrebuiltWorkspaceEventTypeFailed,

Count: int(totalFailed),

})

}

if totalClaimed > 0 {

snapshot.PrebuiltWorkspaces = append(snapshot.PrebuiltWorkspaces, PrebuiltWorkspace{

ID: uuid.New(),

CreatedAt: now,

EventType: PrebuiltWorkspaceEventTypeClaimed,

Count: int(totalClaimed),

})

}

return nil

})

eg.Go(func() error {

tasks, err := CollectTasks(ctx, r.options.Database)

if err != nil {

return xerrors.Errorf("collect tasks telemetry: %w", err)

}

snapshot.Tasks = tasks

return nil

})

eg.Go(func() error {

events, err := CollectTaskEvents(ctx, r.options.Database, createdAfter, now)

if err != nil {

return xerrors.Errorf("collect task events telemetry: %w", err)

}

snapshot.TaskEvents = events

return nil

})

eg.Go(func() error {

summaries, err := r.generateAIBridgeInterceptionsSummaries(ctx)

if err != nil {

return xerrors.Errorf("generate AI Bridge interceptions telemetry summaries: %w", err)

}

snapshot.AIBridgeInterceptionsSummaries = summaries

return nil

})

eg.Go(func() error {

summary, err := r.collectBoundaryUsageSummary(ctx)

if err != nil {

return xerrors.Errorf("collect boundary usage summary: %w", err)

}

// Only send a summary if there was actual usage.

if summary != nil && summary.UniqueUsers > 0 {

snapshot.BoundaryUsageSummary = summary

}

return nil

})

eg.Go(func() error {

chats, err := r.options.Database.GetChatsUpdatedAfter(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get chats updated after: %w", err)

}

snapshot.Chats = make([]Chat, 0, len(chats))

for _, chat := range chats {

snapshot.Chats = append(snapshot.Chats, ConvertChat(chat))

}

return nil

})

eg.Go(func() error {

summaries, err := r.options.Database.GetChatMessageSummariesPerChat(ctx, createdAfter)

if err != nil {

return xerrors.Errorf("get chat message summaries: %w", err)

}

snapshot.ChatMessageSummaries = make([]ChatMessageSummary, 0, len(summaries))

for _, s := range summaries {

snapshot.ChatMessageSummaries = append(snapshot.ChatMessageSummaries, ConvertChatMessageSummary(s))

}

return nil

})

eg.Go(func() error {

configs, err := r.options.Database.GetChatModelConfigsForTelemetry(ctx)

if err != nil {

return xerrors.Errorf("get chat model configs: %w", err)

}

snapshot.ChatModelConfigs = make([]ChatModelConfig, 0, len(configs))

for _, c := range configs {

snapshot.ChatModelConfigs = append(snapshot.ChatModelConfigs, ConvertChatModelConfig(c))

}

return nil

})

eg.Go(func() error {

row, err := r.options.Database.GetChatDiffStatusSummary(ctx)

if err != nil {

return xerrors.Errorf("get chat diff status summary: %w", err)

}

snapshot.ChatDiffStatusSummary = &ChatDiffStatusSummary{

Total: row.Total,

Open: row.Open,

Merged: row.Merged,

Closed: row.Closed,

}

return nil

})

err := eg.Wait()

if err != nil {

return nil, err

}

return snapshot, nil

}

func (r *remoteReporter) generateAIBridgeInterceptionsSummaries(ctx context.Context) ([]AIBridgeInterceptionsSummary, error) {

// Get the current timeframe, which is the previous hour.

now := dbtime.Time(r.options.Clock.Now()).UTC()

endedAtBefore := now.Truncate(time.Hour)

endedAtAfter := endedAtBefore.Add(-1 * time.Hour)

// Note: we don't use a transaction for this function since we do tolerate

// some errors, like duplicate lock rows, and we also calculate

// summaries in parallel.

// Claim the heartbeat lock row for this hour.

err := r.options.Database.InsertTelemetryLock(ctx, database.InsertTelemetryLockParams{

EventType: "aibridge_interceptions_summary",

PeriodEndingAt: endedAtBefore,

})

if database.IsUniqueViolation(err, database.UniqueTelemetryLocksPkey) {

// Another replica has already claimed the lock row for this hour.

r.options.Logger.Debug(ctx, "aibridge interceptions telemetry lock already claimed for this hour by another replica, skipping", slog.F("period_ending_at", endedAtBefore))

return nil, nil

}

if err != nil {

return nil, xerrors.Errorf("insert AI Bridge interceptions telemetry lock (period_ending_at=%q): %w", endedAtBefore, err)

}

// List the summary categories that need to be calculated.

summaryCategories, err := r.options.Database.ListAIBridgeInterceptionsTelemetrySummaries(ctx, database.ListAIBridgeInterceptionsTelemetrySummariesParams{

EndedAtAfter: endedAtAfter, // inclusive

EndedAtBefore: endedAtBefore, // exclusive

})

if err != nil {

return nil, xerrors.Errorf("list AI Bridge interceptions telemetry summaries (startedAtAfter=%q, endedAtBefore=%q): %w", endedAtAfter, endedAtBefore, err)

}

// Calculate and convert the summaries for all categories.

var (

eg, egCtx = errgroup.WithContext(ctx)

mu sync.Mutex

summaries = make([]AIBridgeInterceptionsSummary, 0, len(summaryCategories))

)

for _, category := range summaryCategories {

eg.Go(func() error {

summary, err := r.options.Database.CalculateAIBridgeInterceptionsTelemetrySummary(egCtx, database.CalculateAIBridgeInterceptionsTelemetrySummaryParams{

Provider: category.Provider,

Model: category.Model,

Client: category.Client,

EndedAtAfter: endedAtAfter,

EndedAtBefore: endedAtBefore,

})

if err != nil {

return xerrors.Errorf("calculate AI Bridge interceptions telemetry summary (provider=%q, model=%q, client=%q, startedAtAfter=%q, endedAtBefore=%q): %w", category.Provider, category.Model, category.Client, endedAtAfter, endedAtBefore, err)

}

// Double check that at least one interception was found in the

// timeframe.

if summary.InterceptionCount == 0 {

return nil

}

converted := ConvertAIBridgeInterceptionsSummary(endedAtBefore, category.Provider, category.Model, category.Client, summary)

mu.Lock()

defer mu.Unlock()

summaries = append(summaries, converted)

return nil

})

}

return summaries, eg.Wait()

}

// collectBoundaryUsageSummary collects boundary usage statistics from all

// replicas and resets the stats for the next telemetry period. Returns nil if

// another replica has already collected for this period.

func (r *remoteReporter) collectBoundaryUsageSummary(ctx context.Context) (*BoundaryUsageSummary, error) {

// Use twice the snapshot frequency as the staleness limit to ensure we

// capture data from replicas that may have slightly different flush times.

maxStaleness := r.options.SnapshotFrequency * 2

//nolint:gocritic // This is the actual collection of boundary usage tracking.

boundaryCtx := dbauthz.AsBoundaryUsageTracker(ctx)

// Claim the telemetry lock for this period. Use snapshot frequency so each

// telemetry snapshot period gets exactly one collection.

now := dbtime.Time(r.options.Clock.Now()).UTC()

periodEndingAt := now.Truncate(r.options.SnapshotFrequency)

err := r.options.Database.InsertTelemetryLock(ctx, database.InsertTelemetryLockParams{

EventType: "boundary_usage_summary",

PeriodEndingAt: periodEndingAt,

})

if database.IsUniqueViolation(err, database.UniqueTelemetryLocksPkey) {

r.options.Logger.Debug(ctx, "boundary usage telemetry lock already claimed by another replica, skipping", slog.F("period_ending_at", periodEndingAt))

return nil, nil //nolint:nilnil // This is simple to handle when dealing with telemetry.

}

if err != nil {

return nil, xerrors.Errorf("insert boundary usage telemetry lock (period_ending_at=%q): %w", periodEndingAt, err)

}

var summary database.GetAndResetBoundaryUsageSummaryRow

err = r.options.Database.InTx(func(tx database.Store) error {

// The advisory lock use here ensures a clean transition to the next snapshot by

// preventing replicas from upserting row(s) at the same time as we aggregate and

// delete all rows here.

var txErr error

if txErr = tx.AcquireLock(boundaryCtx, database.LockIDBoundaryUsageStats); txErr != nil {

return txErr

}

summary, txErr = tx.GetAndResetBoundaryUsageSummary(boundaryCtx, maxStaleness.Milliseconds())

return txErr

}, nil)

if err != nil {

return nil, xerrors.Errorf("get and reset boundary usage summary: %w", err)

}

return &BoundaryUsageSummary{

UniqueWorkspaces: summary.UniqueWorkspaces,

UniqueUsers: summary.UniqueUsers,

AllowedRequests: summary.AllowedRequests,

DeniedRequests: summary.DeniedRequests,

PeriodStart: now.Add(-r.options.SnapshotFrequency),

PeriodDurationMilliseconds: r.options.SnapshotFrequency.Milliseconds(),

}, nil

}

func CollectTasks(ctx context.Context, db database.Store) ([]Task, error) {

dbTasks, err := db.ListTasks(ctx, database.ListTasksParams{

OwnerID: uuid.Nil,

OrganizationID: uuid.Nil,

Status: "",

})

if err != nil {

return nil, xerrors.Errorf("list tasks: %w", err)

}

if len(dbTasks) == 0 {

return []Task{}, nil

}

tasks := make([]Task, 0, len(dbTasks))

for _, dbTask := range dbTasks {

tasks = append(tasks, ConvertTask(dbTask))

}

return tasks, nil

}

// buildTaskEvent constructs a TaskEvent from the combined query row.

func buildTaskEvent(

row database.GetTelemetryTaskEventsRow,

createdAfter time.Time,

now time.Time,

) TaskEvent {

event := TaskEvent{

TaskID: row.TaskID.String(),

}

var (

hasStartBuild = row.StartBuildCreatedAt.Valid

isResumed = hasStartBuild && row.StartBuildNumber.Valid && row.StartBuildNumber.Int32 > 1

hasStopBuild = row.StopBuildCreatedAt.Valid

startedAfterStop = hasStartBuild && hasStopBuild && row.StartBuildCreatedAt.Time.After(row.StopBuildCreatedAt.Time)

currentlyPaused = hasStopBuild && !startedAfterStop

)

// Pause-related fields (requires a stop build).

if hasStopBuild {

event.LastPausedAt = &row.StopBuildCreatedAt.Time

switch {

case row.StopBuildReason.Valid && row.StopBuildReason.BuildReason == database.BuildReasonTaskAutoPause:

event.PauseReason = ptr.Ref("auto")

case row.StopBuildReason.Valid && row.StopBuildReason.BuildReason == database.BuildReasonTaskManualPause:

event.PauseReason = ptr.Ref("manual")