15633747409

Committed 13 Jun 2025 11:43AM UTC coverage: 76.585% (+0.3%) from 76.253%

Build # 15633747409

Build Type

Pull #5523

github

Committed by

web-flow

Commit Message

apply suggestions from code review

Co-authored-by: Michel Loiseleur <97035654+mloiseleur@users.noreply.github.com>

Pull Request Pull Request #5523: test(controller): reduce complexity and improve code coverage

Run Details

24 of 61 new or added lines in 1 file covered. (39.34%)

55 existing lines in 4 files now uncovered.

14293 of 18663 relevant lines covered (76.58%)

760.74 hits per line

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

88.96

/controller/controller.go

/*
Copyright 2017 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
        "context"
        "errors"
        "fmt"
        "sync"
        "time"

        "github.com/prometheus/client_golang/prometheus"
        log "github.com/sirupsen/logrus"

        "sigs.k8s.io/external-dns/endpoint"
        "sigs.k8s.io/external-dns/pkg/metrics"
        "sigs.k8s.io/external-dns/plan"
        "sigs.k8s.io/external-dns/provider"
        "sigs.k8s.io/external-dns/registry"
        "sigs.k8s.io/external-dns/source"
)

var (
        registryErrorsTotal = metrics.NewCounterWithOpts(
                prometheus.CounterOpts{
                        Namespace: "external_dns",
                        Subsystem: "registry",
                        Name:      "errors_total",
                        Help:      "Number of Registry errors.",
                },
        )
        sourceErrorsTotal = metrics.NewCounterWithOpts(
                prometheus.CounterOpts{
                        Namespace: "external_dns",
                        Subsystem: "source",
                        Name:      "errors_total",
                        Help:      "Number of Source errors.",
                },
        )
        sourceEndpointsTotal = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "source",
                        Name:      "endpoints_total",
                        Help:      "Number of Endpoints in all sources",
                },
        )
        registryEndpointsTotal = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "registry",
                        Name:      "endpoints_total",
                        Help:      "Number of Endpoints in the registry",
                },
        )
        lastSyncTimestamp = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "last_sync_timestamp_seconds",
                        Help:      "Timestamp of last successful sync with the DNS provider",
                },
        )
        lastReconcileTimestamp = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "last_reconcile_timestamp_seconds",
                        Help:      "Timestamp of last attempted sync with the DNS provider",
                },
        )
        controllerNoChangesTotal = metrics.NewCounterWithOpts(
                prometheus.CounterOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "no_op_runs_total",
                        Help:      "Number of reconcile loops ending up with no changes on the DNS provider side.",
                },
        )
        deprecatedRegistryErrors = metrics.NewCounterWithOpts(
                prometheus.CounterOpts{
                        Subsystem: "registry",
                        Name:      "errors_total",
                        Help:      "Number of Registry errors.",
                },
        )
        deprecatedSourceErrors = metrics.NewCounterWithOpts(
                prometheus.CounterOpts{
                        Subsystem: "source",
                        Name:      "errors_total",
                        Help:      "Number of Source errors.",
                },
        )
        registryARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "registry",
                        Name:      "a_records",
                        Help:      "Number of Registry A records.",
                },
        )
        registryAAAARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "registry",
                        Name:      "aaaa_records",
                        Help:      "Number of Registry AAAA records.",
                },
        )
        sourceARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "source",
                        Name:      "a_records",
                        Help:      "Number of Source A records.",
                },
        )
        sourceAAAARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "source",
                        Name:      "aaaa_records",
                        Help:      "Number of Source AAAA records.",
                },
        )
        verifiedARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "verified_a_records",
                        Help:      "Number of DNS A-records that exists both in source and registry.",
                },
        )
        verifiedAAAARecords = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "verified_aaaa_records",
                        Help:      "Number of DNS AAAA-records that exists both in source and registry.",
                },
        )
        consecutiveSoftErrors = metrics.NewGaugeWithOpts(
                prometheus.GaugeOpts{
                        Namespace: "external_dns",
                        Subsystem: "controller",
                        Name:      "consecutive_soft_errors",
                        Help:      "Number of consecutive soft errors in reconciliation loop.",
                },
        )
)

func init() {
        metrics.RegisterMetric.MustRegister(registryErrorsTotal)
        metrics.RegisterMetric.MustRegister(sourceErrorsTotal)
        metrics.RegisterMetric.MustRegister(sourceEndpointsTotal)
        metrics.RegisterMetric.MustRegister(registryEndpointsTotal)
        metrics.RegisterMetric.MustRegister(lastSyncTimestamp)
        metrics.RegisterMetric.MustRegister(lastReconcileTimestamp)
        metrics.RegisterMetric.MustRegister(deprecatedRegistryErrors)
        metrics.RegisterMetric.MustRegister(deprecatedSourceErrors)
        metrics.RegisterMetric.MustRegister(controllerNoChangesTotal)
        metrics.RegisterMetric.MustRegister(registryARecords)
        metrics.RegisterMetric.MustRegister(registryAAAARecords)
        metrics.RegisterMetric.MustRegister(sourceARecords)
        metrics.RegisterMetric.MustRegister(sourceAAAARecords)
        metrics.RegisterMetric.MustRegister(verifiedARecords)
        metrics.RegisterMetric.MustRegister(verifiedAAAARecords)
        metrics.RegisterMetric.MustRegister(consecutiveSoftErrors)
}

// Controller is responsible for orchestrating the different components.
// It works in the following way:
// * Ask the DNS provider for current list of endpoints.
// * Ask the Source for the desired list of endpoints.
// * Take both lists and calculate a Plan to move current towards desired state.
// * Tell the DNS provider to apply the changes calculated by the Plan.
type Controller struct {
        Source   source.Source
        Registry registry.Registry
        // The policy that defines which changes to DNS records are allowed
        Policy plan.Policy
        // The interval between individual synchronizations
        Interval time.Duration
        // The DomainFilter defines which DNS records to keep or exclude
        DomainFilter endpoint.DomainFilterInterface
        // The nextRunAt used for throttling and batching reconciliation
        nextRunAt time.Time
        // The runAtMutex is for atomic updating of nextRunAt and lastRunAt
        runAtMutex sync.Mutex
        // The lastRunAt used for throttling and batching reconciliation
        lastRunAt time.Time
        // MangedRecordTypes are DNS record types that will be considered for management.
        ManagedRecordTypes []string
        // ExcludeRecordTypes are DNS record types that will be excluded from management.
        ExcludeRecordTypes []string
        // MinEventSyncInterval is used as window for batching events
        MinEventSyncInterval time.Duration
}

// RunOnce runs a single iteration of a reconciliation loop.
func (c *Controller) RunOnce(ctx context.Context) error {
        lastReconcileTimestamp.Gauge.SetToCurrentTime()

        c.runAtMutex.Lock()
        c.lastRunAt = time.Now()
        c.runAtMutex.Unlock()

        records, err := c.Registry.Records(ctx)
        if err != nil {
                registryErrorsTotal.Counter.Inc()
                deprecatedRegistryErrors.Counter.Inc()
                return err
        }

        registryEndpointsTotal.Gauge.Set(float64(len(records)))
        regARecords, regAAAARecords := countAddressRecords(records)
        registryARecords.Gauge.Set(float64(regARecords))
        registryAAAARecords.Gauge.Set(float64(regAAAARecords))
        ctx = context.WithValue(ctx, provider.RecordsContextKey, records)

        endpoints, err := c.Source.Endpoints(ctx)
        if err != nil {
                sourceErrorsTotal.Counter.Inc()
                deprecatedSourceErrors.Counter.Inc()
                return err
        }
        sourceEndpointsTotal.Gauge.Set(float64(len(endpoints)))
        srcARecords, srcAAAARecords := countAddressRecords(endpoints)
        sourceARecords.Gauge.Set(float64(srcARecords))
        sourceAAAARecords.Gauge.Set(float64(srcAAAARecords))
        vARecords, vAAAARecords := countMatchingAddressRecords(endpoints, records)
        verifiedARecords.Gauge.Set(float64(vARecords))
        verifiedAAAARecords.Gauge.Set(float64(vAAAARecords))
        endpoints, err = c.Registry.AdjustEndpoints(endpoints)
        if err != nil {
                return fmt.Errorf("adjusting endpoints: %w", err)
        }
        registryFilter := c.Registry.GetDomainFilter()

        plan := &plan.Plan{
                Policies:       []plan.Policy{c.Policy},
                Current:        records,
                Desired:        endpoints,
                DomainFilter:   endpoint.MatchAllDomainFilters{c.DomainFilter, registryFilter},
                ManagedRecords: c.ManagedRecordTypes,
                ExcludeRecords: c.ExcludeRecordTypes,
                OwnerID:        c.Registry.OwnerID(),
        }

        plan = plan.Calculate()

        if plan.Changes.HasChanges() {
                err = c.Registry.ApplyChanges(ctx, plan.Changes)
                if err != nil {
                        registryErrorsTotal.Counter.Inc()
                        deprecatedRegistryErrors.Counter.Inc()
                        return err
                }
        } else {
                controllerNoChangesTotal.Counter.Inc()
                log.Info("All records are already up to date")
        }

        lastSyncTimestamp.Gauge.SetToCurrentTime()

        return nil
}

func earliest(r time.Time, times ...time.Time) time.Time {
        for _, t := range times {
                if t.Before(r) {
                        r = t
                }
        }
        return r
}

func latest(r time.Time, times ...time.Time) time.Time {
        for _, t := range times {
                if t.After(r) {
                        r = t
                }
        }
        return r
}

// Counts the intersections of A and AAAA records in endpoint and registry.
func countMatchingAddressRecords(endpoints []*endpoint.Endpoint, registryRecords []*endpoint.Endpoint) (int, int) {
        recordsMap := make(map[string]map[string]struct{})
        for _, regRecord := range registryRecords {
                if _, found := recordsMap[regRecord.DNSName]; !found {
                        recordsMap[regRecord.DNSName] = make(map[string]struct{})
                }
                recordsMap[regRecord.DNSName][regRecord.RecordType] = struct{}{}
        }
        aCount := 0
        aaaaCount := 0
        for _, sourceRecord := range endpoints {
                if _, found := recordsMap[sourceRecord.DNSName]; found {
                        if _, found := recordsMap[sourceRecord.DNSName][sourceRecord.RecordType]; found {
                                switch sourceRecord.RecordType {
                                case endpoint.RecordTypeA:
                                        aCount++
                                case endpoint.RecordTypeAAAA:
                                        aaaaCount++
                                }
                        }
                }
        }
        return aCount, aaaaCount
}

func countAddressRecords(endpoints []*endpoint.Endpoint) (int, int) {
        aCount := 0
        aaaaCount := 0
        for _, endPoint := range endpoints {
                switch endPoint.RecordType {
                case endpoint.RecordTypeA:
                        aCount++
                case endpoint.RecordTypeAAAA:
                        aaaaCount++
                }
        }
        return aCount, aaaaCount
}

// ScheduleRunOnce makes sure execution happens at most once per interval.
func (c *Controller) ScheduleRunOnce(now time.Time) {
        c.runAtMutex.Lock()
        defer c.runAtMutex.Unlock()
        c.nextRunAt = latest(
                c.lastRunAt.Add(c.MinEventSyncInterval),
                earliest(
                        now.Add(5*time.Second),
                        c.nextRunAt,
                ),
        )
}

func (c *Controller) ShouldRunOnce(now time.Time) bool {
        c.runAtMutex.Lock()
        defer c.runAtMutex.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()
        var softErrorCount int
        for {
                if c.ShouldRunOnce(time.Now()) {
                        if err := c.RunOnce(ctx); err != nil {
                                if errors.Is(err, provider.SoftError) {
                                        softErrorCount++
                                        consecutiveSoftErrors.Gauge.Set(float64(softErrorCount))
                                        log.Errorf("Failed to do run once: %v (consecutive soft errors: %d)", err, softErrorCount)
                                } else {
                                        log.Fatalf("Failed to do run once: %v", err)
                                }
                        } else {
                                if softErrorCount > 0 {
                                        log.Infof("Reconciliation succeeded after %d consecutive soft errors", softErrorCount)
                                }
                                softErrorCount = 0
                                consecutiveSoftErrors.Gauge.Set(0)
                        }
                }
                select {
                case <-ticker.C:
                case <-ctx.Done():
                        log.Info("Terminating main controller loop")
                        return
                }
        }
}

1	/*
2	Copyright 2017 The Kubernetes Authors.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package controller
18
19	import (
20	"context"
21	"errors"
22	"fmt"
23	"sync"
24	"time"
25
26	"github.com/prometheus/client_golang/prometheus"
27	log "github.com/sirupsen/logrus"
28
29	"sigs.k8s.io/external-dns/endpoint"
30	"sigs.k8s.io/external-dns/pkg/metrics"
31	"sigs.k8s.io/external-dns/plan"
32	"sigs.k8s.io/external-dns/provider"
33	"sigs.k8s.io/external-dns/registry"
34	"sigs.k8s.io/external-dns/source"
35	)
36
37	var (
38	registryErrorsTotal = metrics.NewCounterWithOpts(
39	prometheus.CounterOpts{
40	Namespace: "external_dns",
41	Subsystem: "registry",
42	Name: "errors_total",
43	Help: "Number of Registry errors.",
44	},
45	)
46	sourceErrorsTotal = metrics.NewCounterWithOpts(
47	prometheus.CounterOpts{
48	Namespace: "external_dns",
49	Subsystem: "source",
50	Name: "errors_total",
51	Help: "Number of Source errors.",
52	},
53	)
54	sourceEndpointsTotal = metrics.NewGaugeWithOpts(
55	prometheus.GaugeOpts{
56	Namespace: "external_dns",
57	Subsystem: "source",
58	Name: "endpoints_total",
59	Help: "Number of Endpoints in all sources",
60	},
61	)
62	registryEndpointsTotal = metrics.NewGaugeWithOpts(
63	prometheus.GaugeOpts{
64	Namespace: "external_dns",
65	Subsystem: "registry",
66	Name: "endpoints_total",
67	Help: "Number of Endpoints in the registry",
68	},
69	)
70	lastSyncTimestamp = metrics.NewGaugeWithOpts(
71	prometheus.GaugeOpts{
72	Namespace: "external_dns",
73	Subsystem: "controller",
74	Name: "last_sync_timestamp_seconds",
75	Help: "Timestamp of last successful sync with the DNS provider",
76	},
77	)
78	lastReconcileTimestamp = metrics.NewGaugeWithOpts(
79	prometheus.GaugeOpts{
80	Namespace: "external_dns",
81	Subsystem: "controller",
82	Name: "last_reconcile_timestamp_seconds",
83	Help: "Timestamp of last attempted sync with the DNS provider",
84	},
85	)
86	controllerNoChangesTotal = metrics.NewCounterWithOpts(
87	prometheus.CounterOpts{
88	Namespace: "external_dns",
89	Subsystem: "controller",
90	Name: "no_op_runs_total",
91	Help: "Number of reconcile loops ending up with no changes on the DNS provider side.",
92	},
93	)
94	deprecatedRegistryErrors = metrics.NewCounterWithOpts(
95	prometheus.CounterOpts{
96	Subsystem: "registry",
97	Name: "errors_total",
98	Help: "Number of Registry errors.",
99	},
100	)
101	deprecatedSourceErrors = metrics.NewCounterWithOpts(
102	prometheus.CounterOpts{
103	Subsystem: "source",
104	Name: "errors_total",
105	Help: "Number of Source errors.",
106	},
107	)
108	registryARecords = metrics.NewGaugeWithOpts(
109	prometheus.GaugeOpts{
110	Namespace: "external_dns",
111	Subsystem: "registry",
112	Name: "a_records",
113	Help: "Number of Registry A records.",
114	},
115	)
116	registryAAAARecords = metrics.NewGaugeWithOpts(
117	prometheus.GaugeOpts{
118	Namespace: "external_dns",
119	Subsystem: "registry",
120	Name: "aaaa_records",
121	Help: "Number of Registry AAAA records.",
122	},
123	)
124	sourceARecords = metrics.NewGaugeWithOpts(
125	prometheus.GaugeOpts{
126	Namespace: "external_dns",
127	Subsystem: "source",
128	Name: "a_records",
129	Help: "Number of Source A records.",
130	},
131	)
132	sourceAAAARecords = metrics.NewGaugeWithOpts(
133	prometheus.GaugeOpts{
134	Namespace: "external_dns",
135	Subsystem: "source",
136	Name: "aaaa_records",
137	Help: "Number of Source AAAA records.",
138	},
139	)
140	verifiedARecords = metrics.NewGaugeWithOpts(
141	prometheus.GaugeOpts{
142	Namespace: "external_dns",
143	Subsystem: "controller",
144	Name: "verified_a_records",
145	Help: "Number of DNS A-records that exists both in source and registry.",
146	},
147	)
148	verifiedAAAARecords = metrics.NewGaugeWithOpts(
149	prometheus.GaugeOpts{	1✔
150	Namespace: "external_dns",	1✔
151	Subsystem: "controller",	1✔
152	Name: "verified_aaaa_records",	1✔
153	Help: "Number of DNS AAAA-records that exists both in source and registry.",	1✔
154	},	1✔
155	)	1✔
156	consecutiveSoftErrors = metrics.NewGaugeWithOpts(	1✔
157	prometheus.GaugeOpts{	1✔
158	Namespace: "external_dns",	1✔
159	Subsystem: "controller",	1✔
160	Name: "consecutive_soft_errors",	1✔
161	Help: "Number of consecutive soft errors in reconciliation loop.",	1✔
162	},	1✔
163	)	1✔
164	)	1✔
165		1✔
166	func init() {
167	metrics.RegisterMetric.MustRegister(registryErrorsTotal)
168	metrics.RegisterMetric.MustRegister(sourceErrorsTotal)
169	metrics.RegisterMetric.MustRegister(sourceEndpointsTotal)
170	metrics.RegisterMetric.MustRegister(registryEndpointsTotal)
171	metrics.RegisterMetric.MustRegister(lastSyncTimestamp)
172	metrics.RegisterMetric.MustRegister(lastReconcileTimestamp)
173	metrics.RegisterMetric.MustRegister(deprecatedRegistryErrors)
174	metrics.RegisterMetric.MustRegister(deprecatedSourceErrors)
175	metrics.RegisterMetric.MustRegister(controllerNoChangesTotal)
176	metrics.RegisterMetric.MustRegister(registryARecords)
177	metrics.RegisterMetric.MustRegister(registryAAAARecords)
178	metrics.RegisterMetric.MustRegister(sourceARecords)
179	metrics.RegisterMetric.MustRegister(sourceAAAARecords)
180	metrics.RegisterMetric.MustRegister(verifiedARecords)
181	metrics.RegisterMetric.MustRegister(verifiedAAAARecords)
182	metrics.RegisterMetric.MustRegister(consecutiveSoftErrors)
183	}
184
185	// Controller is responsible for orchestrating the different components.
186	// It works in the following way:
187	// * Ask the DNS provider for current list of endpoints.
188	// * Ask the Source for the desired list of endpoints.
189	// * Take both lists and calculate a Plan to move current towards desired state.
190	// * Tell the DNS provider to apply the changes calculated by the Plan.
191	type Controller struct {
192	Source source.Source
193	Registry registry.Registry
194	// The policy that defines which changes to DNS records are allowed
195	Policy plan.Policy
196	// The interval between individual synchronizations
197	Interval time.Duration	16✔
198	// The DomainFilter defines which DNS records to keep or exclude	16✔
199	DomainFilter endpoint.DomainFilterInterface	16✔
200	// The nextRunAt used for throttling and batching reconciliation	16✔
201	nextRunAt time.Time	16✔
202	// The runAtMutex is for atomic updating of nextRunAt and lastRunAt	16✔
203	runAtMutex sync.Mutex	16✔
204	// The lastRunAt used for throttling and batching reconciliation	16✔
205	lastRunAt time.Time	16✔
206	// MangedRecordTypes are DNS record types that will be considered for management.	16✔
207	ManagedRecordTypes []string	19✔
208	// ExcludeRecordTypes are DNS record types that will be excluded from management.	3✔
209	ExcludeRecordTypes []string	3✔
210	// MinEventSyncInterval is used as window for batching events	3✔
211	MinEventSyncInterval time.Duration	3✔
212	}
213		13✔
214	// RunOnce runs a single iteration of a reconciliation loop.	13✔
215	func (c *Controller) RunOnce(ctx context.Context) error {	13✔
216	lastReconcileTimestamp.Gauge.SetToCurrentTime()	13✔
217		13✔
218	c.runAtMutex.Lock()	13✔
219	c.lastRunAt = time.Now()	13✔
220	c.runAtMutex.Unlock()	13✔
UNCOV 221		×
UNCOV 222	records, err := c.Registry.Records(ctx)	×
UNCOV 223	if err != nil {	×
UNCOV 224	registryErrorsTotal.Counter.Inc()	×
225	deprecatedRegistryErrors.Counter.Inc()
226	return err	13✔
227	}	13✔
228		13✔
229	registryEndpointsTotal.Gauge.Set(float64(len(records)))	13✔
230	regARecords, regAAAARecords := countAddressRecords(records)	13✔
231	registryARecords.Gauge.Set(float64(regARecords))	13✔
232	registryAAAARecords.Gauge.Set(float64(regAAAARecords))	13✔
233	ctx = context.WithValue(ctx, provider.RecordsContextKey, records)	13✔
234		13✔
235	endpoints, err := c.Source.Endpoints(ctx)	13✔
UNCOV 236	if err != nil {	×
237	sourceErrorsTotal.Counter.Inc()	×
238	deprecatedSourceErrors.Counter.Inc()	13✔
239	return err	13✔
240	}	13✔
241	sourceEndpointsTotal.Gauge.Set(float64(len(endpoints)))	13✔
242	srcARecords, srcAAAARecords := countAddressRecords(endpoints)	13✔
243	sourceARecords.Gauge.Set(float64(srcARecords))	13✔
244	sourceAAAARecords.Gauge.Set(float64(srcAAAARecords))	13✔
245	vARecords, vAAAARecords := countMatchingAddressRecords(endpoints, records)	13✔
246	verifiedARecords.Gauge.Set(float64(vARecords))	13✔
247	verifiedAAAARecords.Gauge.Set(float64(vAAAARecords))	13✔
248	endpoints, err = c.Registry.AdjustEndpoints(endpoints)	13✔
249	if err != nil {	13✔
250	return fmt.Errorf("adjusting endpoints: %w", err)	13✔
251	}	13✔
252	registryFilter := c.Registry.GetDomainFilter()	23✔
253		10✔
254	plan := &plan.Plan{	10✔
UNCOV 255	Policies: []plan.Policy{c.Policy},	×
UNCOV 256	Current: records,	×
UNCOV 257	Desired: endpoints,	×
UNCOV 258	DomainFilter: endpoint.MatchAllDomainFilters{c.DomainFilter, registryFilter},	×
259	ManagedRecords: c.ManagedRecordTypes,	3✔
260	ExcludeRecords: c.ExcludeRecordTypes,	3✔
261	OwnerID: c.Registry.OwnerID(),	3✔
262	}	3✔
263
264	plan = plan.Calculate()	13✔
265		13✔
266	if plan.Changes.HasChanges() {	13✔
267	err = c.Registry.ApplyChanges(ctx, plan.Changes)
268	if err != nil {
269	registryErrorsTotal.Counter.Inc()	4✔
270	deprecatedRegistryErrors.Counter.Inc()	8✔
271	return err	4✔
272	}	×
UNCOV 273	} else {	×
274	controllerNoChangesTotal.Counter.Inc()
275	log.Info("All records are already up to date")	4✔
276	}
277
278	lastSyncTimestamp.Gauge.SetToCurrentTime()	4✔
279		8✔
280	return nil	4✔
UNCOV 281	}	×
UNCOV 282		×
283	func earliest(r time.Time, times ...time.Time) time.Time {
284	for _, t := range times {	4✔
285	if t.Before(r) {
286	r = t
287	}
288	}	13✔
289	return r	13✔
290	}	5,796✔
291		11,566✔
292	func latest(r time.Time, times ...time.Time) time.Time {	5,783✔
293	for _, t := range times {	5,783✔
294	if t.After(r) {	5,783✔
295	r = t
296	}
297	}	978✔
298	return r	1,520✔
299	}	1,110✔
300		555✔
301	// Counts the intersections of A and AAAA records in endpoint and registry.	555✔
302	func countMatchingAddressRecords(endpoints []endpoint.Endpoint, registryRecords []endpoint.Endpoint) (int, int) {
303	recordsMap := make(map[string]map[string]struct{})
304	for _, regRecord := range registryRecords {
305	if _, found := recordsMap[regRecord.DNSName]; !found {	117✔
306	recordsMap[regRecord.DNSName] = make(map[string]struct{})	104✔
307	}	104✔
308	recordsMap[regRecord.DNSName][regRecord.RecordType] = struct{}{}
309	}
310	aCount := 0
311	aaaaCount := 0
312	for _, sourceRecord := range endpoints {
313	if _, found := recordsMap[sourceRecord.DNSName]; found {	26✔
314	if _, found := recordsMap[sourceRecord.DNSName][sourceRecord.RecordType]; found {	26✔
315	switch sourceRecord.RecordType {	6,774✔
316	case endpoint.RecordTypeA:	6,748✔
317	aCount++	6,748✔
318	case endpoint.RecordTypeAAAA:
319	aaaaCount++	234✔
320	}	208✔
321	}	208✔
322	}
323	}
324	return aCount, aaaaCount
325	}	4✔
326		4✔
327	func countAddressRecords(endpoints []*endpoint.Endpoint) (int, int) {	4✔
328	aCount := 0	4✔
329	aaaaCount := 0	4✔
330	for _, endPoint := range endpoints {	4✔
331	switch endPoint.RecordType {	4✔
332	case endpoint.RecordTypeA:	4✔
333	aCount++	4✔
334	case endpoint.RecordTypeAAAA:	4✔
335	aaaaCount++	4✔
336	}
337	}	16✔
338	return aCount, aaaaCount	16✔
339	}	16✔
340		23✔
341	// ScheduleRunOnce makes sure execution happens at most once per interval.	7✔
342	func (c *Controller) ScheduleRunOnce(now time.Time) {	7✔
343	c.runAtMutex.Lock()	9✔
344	defer c.runAtMutex.Unlock()	9✔
345	c.nextRunAt = latest(
346	c.lastRunAt.Add(c.MinEventSyncInterval),
347	earliest(
348	now.Add(5*time.Second),	2✔
349	c.nextRunAt,	2✔
350	),	2✔
351	)	2✔
352	}	7✔
353		10✔
354	func (c *Controller) ShouldRunOnce(now time.Time) bool {	8✔
355	c.runAtMutex.Lock()	6✔
356	defer c.runAtMutex.Unlock()	3✔
357	if now.Before(c.nextRunAt) {	3✔
358	return false	3✔
359	}	3✔
UNCOV 360	c.nextRunAt = now.Add(c.Interval)	×
UNCOV 361	return true	×
362	}	2✔
363		2✔
UNCOV 364	// Run runs RunOnce in a loop with a delay until context is canceled	×
UNCOV 365	func (c *Controller) Run(ctx context.Context) {	×
366	ticker := time.NewTicker(time.Second)	2✔
367	defer ticker.Stop()	2✔
368	var softErrorCount int
369	for {
370	if c.ShouldRunOnce(time.Now()) {	5✔
371	if err := c.RunOnce(ctx); err != nil {	3✔
372	if errors.Is(err, provider.SoftError) {	2✔
373	softErrorCount++	2✔
374	consecutiveSoftErrors.Gauge.Set(float64(softErrorCount))	2✔
375	log.Errorf("Failed to do run once: %v (consecutive soft errors: %d)", err, softErrorCount)
376	} else {
377	log.Fatalf("Failed to do run once: %v", err)
378	}
379	} else {
380	if softErrorCount > 0 {
381	log.Infof("Reconciliation succeeded after %d consecutive soft errors", softErrorCount)
382	}
383	softErrorCount = 0
384	consecutiveSoftErrors.Gauge.Set(0)
385	}
386	}
387	select {
388	case <-ticker.C:
389	case <-ctx.Done():
390	log.Info("Terminating main controller loop")
391	return
392	}
393	}
394	}

kubernetes-sigs / external-dns / 15633747409

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