Method run

pkg/module/metrics/metrics_module.go:269–342

(newCtx context.Context)

}

func (m *Module) run(newCtx context.Context) {
	if m.isRunning {
		m.l.Warn("Metric module is already running. Cannot start again.")
		return
	}

	cbFunc := pubsub.CallBackFunc(m.PodCallBackFn)
	m.pubsubPodSub = m.pubsub.Subscribe(common.PubSubPods, &cbFunc)

	m.wg.Add(1)
	go func() {
		m.Lock()
		m.isRunning = true
		m.ctx = newCtx
		m.Unlock()

		evReader := m.enricher.ExportReader()
		for {
			ev := evReader.NextFollow(newCtx)
			if ev == nil {
				break
			}

			switch ev.Event.(type) {
			case *flow.Flow:
				m.RLock()
				f := ev.Event.(*flow.Flow)
				m.l.Debug("converted flow object", zap.Any("flow l4", f.IP))
				for _, metricObj := range m.registry {
					metricObj.ProcessFlow(f)
				}
				m.RUnlock()
			case *flow.LostEvent:
				ev := ev.Event.(*flow.LostEvent)
				// the number of lost events == the size of the ring buffer initialized.
				metrics.LostEventsCounter.WithLabelValues(utils.EnricherRing, string(metricModuleReq)).Add(float64(ev.NumEventsLost))
			default:
				m.l.Warn("Unknown event type", zap.Any("event", ev))
			}
		}

		err := evReader.Close()
		if err != nil {
			m.l.Error("Error closing the event reader", zap.Error(err))
		}
		m.Lock()
		m.isRunning = false
		m.ctx = nil
		m.Unlock()

		m.wg.Done()
	}()

	m.wg.Add(1)
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {