Method Start

service/netquery/module_api.go:182–260

()

}

func (nq *NetQuery) Start() error {
	nq.mgr.Go("netquery connection feed listener", func(ctx *mgr.WorkerCtx) error {
		sub, err := nq.db.Subscribe(query.New("network:"))
		if err != nil {
			return fmt.Errorf("failed to subscribe to network tree: %w", err)
		}
		defer close(nq.feed)
		defer func() {
			_ = sub.Cancel()
		}()

		for {
			select {
			case <-ctx.Done():
				return nil
			case rec, ok := <-sub.Feed:
				if !ok {
					return nil
				}

				conn, ok := rec.(*network.Connection)
				if !ok {
					// This is fine as we also receive process updates on
					// this channel.
					continue
				}

				nq.feed <- conn
			}
		}
	})

	nq.mgr.Go("netquery connection feed handler", func(ctx *mgr.WorkerCtx) error {
		nq.mng.HandleFeed(ctx.Ctx(), nq.feed)
		return nil
	})

	nq.mgr.Go("netquery live db cleaner", func(ctx *mgr.WorkerCtx) error {
		for {
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(10 * time.Second):
				threshold := time.Now().Add(-network.DeleteConnsAfterEndedThreshold)
				count, err := nq.Store.Cleanup(ctx.Ctx(), threshold)
				if err != nil {
					log.Errorf("netquery: failed to removed old connections from live db: %s", err)
				} else {
					log.Tracef("netquery: successfully removed %d old connections from live db that ended before %s", count, threshold)
				}
			}
		}
	})

	nq.mgr.Delay("network history cleaner delay", 10*time.Minute, func(w *mgr.WorkerCtx) error {
		return nq.Store.CleanupHistory(w.Ctx())
	}).Repeat(1 * time.Hour)