Method Start

agent/proto/resourcesmonitor/resources_monitor.go:37–93

(ctx context.Context)

}

func (m *monitor) Start(ctx context.Context) error {
	m.clock.TickerFunc(ctx, time.Duration(m.config.Config.CollectionIntervalSeconds)*time.Second, func() error {
		datapoint := Datapoint{
			CollectedAt: m.clock.Now(),
			Volumes:     make([]*VolumeDatapoint, 0, len(m.config.Volumes)),
		}

		if m.config.Memory != nil && m.config.Memory.Enabled {
			memTotal, memUsed, err := m.resourcesFetcher.FetchMemory()
			if err != nil {
				m.logger.Error(ctx, "failed to fetch memory", slog.Error(err))
			} else {
				datapoint.Memory = &MemoryDatapoint{
					Total: memTotal,
					Used:  memUsed,
				}
			}
		}

		for _, volume := range m.config.Volumes {
			if !volume.Enabled {
				continue
			}

			volTotal, volUsed, err := m.resourcesFetcher.FetchVolume(volume.Path)
			if err != nil {
				m.logger.Error(ctx, "failed to fetch volume", slog.Error(err))
				continue
			}

			datapoint.Volumes = append(datapoint.Volumes, &VolumeDatapoint{
				Path:  volume.Path,
				Total: volTotal,
				Used:  volUsed,
			})
		}

		m.queue.Push(datapoint)

		if m.queue.IsFull() {
			_, err := m.datapointsPusher.PushResourcesMonitoringUsage(ctx, &proto.PushResourcesMonitoringUsageRequest{
				Datapoints: m.queue.ItemsAsProto(),
			})
			if err != nil {
				// We don't want to stop the monitoring if we fail to push the datapoints
				// to the server. We just log the error and continue.
				// The queue will anyway remove the oldest datapoint and add the new one.
				m.logger.Error(ctx, "failed to push resources monitoring usage", slog.Error(err))
				return nil
			}
		}

		return nil
	}, "resources_monitor")

	return nil
}