14813914835

Committed 03 May 2025 07:23PM UTC coverage: 60.792% (+3.1%) from 57.655%

Build # 14813914835

Build Type

push

github

Committed by

sapslaj

Commit Message

feat: "http" provider

A simple provider that exposes a JSON list representation of endpoints
at the `/endpoints` HTTP endpoint. This is similar to the "http" Lua
module but for pull-based systems.

Coverage Stats

5 of 42 new or added lines in 2 files covered. (11.9%)

51 existing lines in 4 files now uncovered.

1090 of 1793 relevant lines covered (60.79%)

5.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.83

/controller/controller.go

package controller

import (
        "context"
        "sync"
        "time"

        "go.uber.org/multierr"
        "go.uber.org/zap"

        "github.com/sapslaj/zonepop/config/configtypes"
        "github.com/sapslaj/zonepop/endpoint"
        "github.com/sapslaj/zonepop/provider"
        "github.com/sapslaj/zonepop/source"
)

type Controller struct {
        Sources   []source.NamedSource
        Providers []provider.NamedProvider
        // The interval between individual synchronizations
        Interval time.Duration
        // Logger instance
        Logger *zap.Logger
        // The nextRunAt used for throttling and batching reconciliation
        nextRunAt time.Time
        // The nextRunAtMux is for atomic updating of nextRunAt
        nextRunAtMux sync.Mutex
}

// RunOnce runs a single iteration of a reconciliation loop.
func (c *Controller) RunOnce(ctx context.Context) error {
        var errors error
        start := time.Now()
        // Hack to force the labels to always show up
        MetricRuns.WithLabelValues("success").Add(0)
        MetricRuns.WithLabelValues("errored").Add(0)
        defer func() {
                status := "success"
                if errors != nil {
                        status = "errored"
                }
                MetricRuns.WithLabelValues(status).Inc()
                MetricLastRunTimestamp.SetToCurrentTime()
                duration := time.Since(start)
                MetricLastRunDurationSeconds.Set(duration.Seconds())
                MetricRunDurationSeconds.Observe(duration.Seconds())
        }()
        logger := c.Logger.Sugar()
        endpoints := make([]*endpoint.Endpoint, 0)
        for _, s := range c.Sources {
                e, err := s.Source.Endpoints(ctx)
                if err != nil {
                        logger.Errorw(
                                "error getting endpoints from source",
                                "source", s.Name,
                                "err", err,
                        )
                        errors = multierr.Append(errors, err)
                        MetricSourceUp.WithLabelValues(s.Name).Set(0)
                } else {
                        MetricSourceUp.WithLabelValues(s.Name).Set(1)
                }
                MetricEndpoints.WithLabelValues(s.Name).Set(float64(len(e)))
                for i := range e {
                        if e[i].SourceProperties == nil {
                                e[i].SourceProperties = map[string]any{}
                        }
                        e[i].SourceProperties["source"] = s.Name
                }
                endpoints = append(endpoints, e...)
        }
        if errors != nil {
                // Bail early, and set all providers to down
                for _, p := range c.Providers {
                        MetricProviderUp.WithLabelValues(p.Name).Set(0)
                }
                return errors
        }
        for _, endpoint := range endpoints {
                logger.Infow(
                        "registered new endpoint",
                        "hostname", endpoint.Hostname,
                        "ipv4", endpoint.IPv4s,
                        "ipv6", endpoint.IPv6s,
                        "ttl", endpoint.RecordTTL,
                        "source_properties", endpoint.SourceProperties,
                        "provider_properties", endpoint.ProviderProperties,
                )
        }
        dryRun, ok := ctx.Value(configtypes.DryRunContextKey).(bool)
        if !ok {
                dryRun = false
        }
        if !dryRun {
                for _, p := range c.Providers {
                        err := p.Provider.UpdateEndpoints(ctx, endpoints)
                        if err != nil {
                                logger.Errorw(
                                        "error updating endpoints with provider",
                                        "provider", p.Name,
                                        "err", err,
                                )
                                errors = multierr.Append(errors, err)
                                MetricProviderUp.WithLabelValues(p.Name).Set(0)
                        } else {
                                MetricProviderUp.WithLabelValues(p.Name).Set(1)
                        }
                }
        }
        return errors
}

// ScheduleRunOnce makes sure execution happens at most once per interval.
func (c *Controller) ScheduleRunOnce(now time.Time) {
        c.nextRunAtMux.Lock()
        defer c.nextRunAtMux.Unlock()
}

func (c *Controller) ShouldRunOnce(now time.Time) bool {
        c.nextRunAtMux.Lock()
        defer c.nextRunAtMux.Unlock()
        if now.Before(c.nextRunAt) {
                return false
        }
        c.nextRunAt = now.Add(c.Interval)
        return true
}

// Run runs RunOnce in a loop with a delay until context is canceled.
func (c *Controller) Run(ctx context.Context) {
        ticker := time.NewTicker(time.Second)
        defer ticker.Stop()
        for {
                if c.ShouldRunOnce(time.Now()) {
                        if err := c.RunOnce(ctx); err != nil {
                                c.Logger.Sugar().Errorf("controller.Run error: %v", err)
                        }
                }
                select {
                case <-ticker.C:
                case <-ctx.Done():
                        c.Logger.Info("Terminating main controller loop")
                        return
                }
        }
}

1	package controller
2
3	import (
4	"context"
5	"sync"
6	"time"
7
8	"go.uber.org/multierr"
9	"go.uber.org/zap"
10
11	"github.com/sapslaj/zonepop/config/configtypes"
12	"github.com/sapslaj/zonepop/endpoint"
13	"github.com/sapslaj/zonepop/provider"
14	"github.com/sapslaj/zonepop/source"
15	)
16
17	type Controller struct {
18	Sources []source.NamedSource
19	Providers []provider.NamedProvider
20	// The interval between individual synchronizations
21	Interval time.Duration
22	// Logger instance
23	Logger *zap.Logger
24	// The nextRunAt used for throttling and batching reconciliation
25	nextRunAt time.Time
26	// The nextRunAtMux is for atomic updating of nextRunAt
27	nextRunAtMux sync.Mutex
28	}
29
30	// RunOnce runs a single iteration of a reconciliation loop.
31	func (c *Controller) RunOnce(ctx context.Context) error {	4✔
32	var errors error	4✔
33	start := time.Now()	4✔
34	// Hack to force the labels to always show up	4✔
35	MetricRuns.WithLabelValues("success").Add(0)	4✔
36	MetricRuns.WithLabelValues("errored").Add(0)	4✔
37	defer func() {	8✔
38	status := "success"	4✔
39	if errors != nil {	6✔
40	status = "errored"	2✔
41	}	2✔
42	MetricRuns.WithLabelValues(status).Inc()	4✔
43	MetricLastRunTimestamp.SetToCurrentTime()	4✔
44	duration := time.Since(start)	4✔
45	MetricLastRunDurationSeconds.Set(duration.Seconds())	4✔
46	MetricRunDurationSeconds.Observe(duration.Seconds())	4✔
47	}()
48	logger := c.Logger.Sugar()	4✔
49	endpoints := make([]*endpoint.Endpoint, 0)	4✔
50	for _, s := range c.Sources {	9✔
51	e, err := s.Source.Endpoints(ctx)	5✔
52	if err != nil {	6✔
53	logger.Errorw(	1✔
54	"error getting endpoints from source",	1✔
55	"source", s.Name,	1✔
56	"err", err,	1✔
57	)	1✔
58	errors = multierr.Append(errors, err)	1✔
59	MetricSourceUp.WithLabelValues(s.Name).Set(0)	1✔
60	} else {	5✔
61	MetricSourceUp.WithLabelValues(s.Name).Set(1)	4✔
62	}	4✔
63	MetricEndpoints.WithLabelValues(s.Name).Set(float64(len(e)))	5✔
64	for i := range e {	9✔
65	if e[i].SourceProperties == nil {	7✔
66	e[i].SourceProperties = map[string]any{}	3✔
67	}	3✔
68	e[i].SourceProperties["source"] = s.Name	4✔
69	}
70	endpoints = append(endpoints, e...)	5✔
71	}
72	if errors != nil {	5✔
73	// Bail early, and set all providers to down	1✔
74	for _, p := range c.Providers {	3✔
75	MetricProviderUp.WithLabelValues(p.Name).Set(0)	2✔
76	}	2✔
77	return errors	1✔
78	}
79	for _, endpoint := range endpoints {	6✔
80	logger.Infow(	3✔
81	"registered new endpoint",	3✔
82	"hostname", endpoint.Hostname,	3✔
83	"ipv4", endpoint.IPv4s,	3✔
84	"ipv6", endpoint.IPv6s,	3✔
85	"ttl", endpoint.RecordTTL,	3✔
86	"source_properties", endpoint.SourceProperties,	3✔
87	"provider_properties", endpoint.ProviderProperties,	3✔
88	)	3✔
89	}	3✔
90	dryRun, ok := ctx.Value(configtypes.DryRunContextKey).(bool)	3✔
91	if !ok {	5✔
92	dryRun = false	2✔
93	}	2✔
94	if !dryRun {	5✔
95	for _, p := range c.Providers {	5✔
96	err := p.Provider.UpdateEndpoints(ctx, endpoints)	3✔
97	if err != nil {	4✔
98	logger.Errorw(	1✔
99	"error updating endpoints with provider",	1✔
100	"provider", p.Name,	1✔
101	"err", err,	1✔
102	)	1✔
103	errors = multierr.Append(errors, err)	1✔
104	MetricProviderUp.WithLabelValues(p.Name).Set(0)	1✔
105	} else {	3✔
106	MetricProviderUp.WithLabelValues(p.Name).Set(1)	2✔
107	}	2✔
108	}
109	}
110	return errors	3✔
111	}
112
113	// ScheduleRunOnce makes sure execution happens at most once per interval.
UNCOV 114	func (c *Controller) ScheduleRunOnce(now time.Time) {	×
UNCOV 115	c.nextRunAtMux.Lock()	×
UNCOV 116	defer c.nextRunAtMux.Unlock()	×
UNCOV 117	}	×
118
119	func (c *Controller) ShouldRunOnce(now time.Time) bool {	4✔
120	c.nextRunAtMux.Lock()	4✔
121	defer c.nextRunAtMux.Unlock()	4✔
122	if now.Before(c.nextRunAt) {	6✔
123	return false	2✔
124	}	2✔
125	c.nextRunAt = now.Add(c.Interval)	2✔
126	return true	2✔
127	}
128
129	// Run runs RunOnce in a loop with a delay until context is canceled.
130	func (c *Controller) Run(ctx context.Context) {	×
131	ticker := time.NewTicker(time.Second)	×
UNCOV 132	defer ticker.Stop()	×
133	for {	×
134	if c.ShouldRunOnce(time.Now()) {	×
135	if err := c.RunOnce(ctx); err != nil {	×
136	c.Logger.Sugar().Errorf("controller.Run error: %v", err)	×
137	}	×
138	}
UNCOV 139	select {	×
UNCOV 140	case <-ticker.C:	×
UNCOV 141	case <-ctx.Done():	×
UNCOV 142	c.Logger.Info("Terminating main controller loop")	×
UNCOV 143	return	×
144	}
145	}
146	}

sapslaj / zonepop / 14813914835

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous