13654368756

Committed 04 Mar 2025 01:05PM UTC coverage: 70.721% (+0.2%) from 70.57%

Build # 13654368756

Build Type

push

github

Committed by

web-flow

Commit Message

chore(docs): generate docs/monitoring/metrics.md file (#5117)

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): generate docs/monitoring/metrics.md file

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

* chore(docs): gen... (continued)

Run Details

184 of 232 new or added lines in 9 files covered. (79.31%)

1 existing line in 1 file now uncovered.

14183 of 20055 relevant lines covered (70.72%)

663.2 hits per line

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

85.26

/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.",
                },
        )
)

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)
}

// 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()
        for {
                if c.ShouldRunOnce(time.Now()) {
                        if err := c.RunOnce(ctx); err != nil {
                                if errors.Is(err, provider.SoftError) {
                                        log.Errorf("Failed to do run once: %v", err)
                                } else {
                                        log.Fatalf("Failed to do run once: %v", err)
                                }
                        }
                }
                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{
150	Namespace: "external_dns",
151	Subsystem: "controller",
152	Name: "verified_aaaa_records",
153	Help: "Number of DNS AAAA-records that exists both in source and registry.",
154	},
155	)
156	)
157
158	func init() {	1✔
159	metrics.RegisterMetric.MustRegister(registryErrorsTotal)	1✔
160	metrics.RegisterMetric.MustRegister(sourceErrorsTotal)	1✔
161	metrics.RegisterMetric.MustRegister(sourceEndpointsTotal)	1✔
162	metrics.RegisterMetric.MustRegister(registryEndpointsTotal)	1✔
163	metrics.RegisterMetric.MustRegister(lastSyncTimestamp)	1✔
164	metrics.RegisterMetric.MustRegister(lastReconcileTimestamp)	1✔
165	metrics.RegisterMetric.MustRegister(deprecatedRegistryErrors)	1✔
166	metrics.RegisterMetric.MustRegister(deprecatedSourceErrors)	1✔
167	metrics.RegisterMetric.MustRegister(controllerNoChangesTotal)	1✔
168	metrics.RegisterMetric.MustRegister(registryARecords)	1✔
169	metrics.RegisterMetric.MustRegister(registryAAAARecords)	1✔
170	metrics.RegisterMetric.MustRegister(sourceARecords)	1✔
171	metrics.RegisterMetric.MustRegister(sourceAAAARecords)	1✔
172	metrics.RegisterMetric.MustRegister(verifiedARecords)	1✔
173	metrics.RegisterMetric.MustRegister(verifiedAAAARecords)	1✔
174	}	1✔
175
176	// Controller is responsible for orchestrating the different components.
177	// It works in the following way:
178	// * Ask the DNS provider for current list of endpoints.
179	// * Ask the Source for the desired list of endpoints.
180	// * Take both lists and calculate a Plan to move current towards desired state.
181	// * Tell the DNS provider to apply the changes calculated by the Plan.
182	type Controller struct {
183	Source source.Source
184	Registry registry.Registry
185	// The policy that defines which changes to DNS records are allowed
186	Policy plan.Policy
187	// The interval between individual synchronizations
188	Interval time.Duration
189	// The DomainFilter defines which DNS records to keep or exclude
190	DomainFilter endpoint.DomainFilterInterface
191	// The nextRunAt used for throttling and batching reconciliation
192	nextRunAt time.Time
193	// The runAtMutex is for atomic updating of nextRunAt and lastRunAt
194	runAtMutex sync.Mutex
195	// The lastRunAt used for throttling and batching reconciliation
196	lastRunAt time.Time
197	// MangedRecordTypes are DNS record types that will be considered for management.
198	ManagedRecordTypes []string
199	// ExcludeRecordTypes are DNS record types that will be excluded from management.
200	ExcludeRecordTypes []string
201	// MinEventSyncInterval is used as window for batching events
202	MinEventSyncInterval time.Duration
203	}
204
205	// RunOnce runs a single iteration of a reconciliation loop.
206	func (c *Controller) RunOnce(ctx context.Context) error {	12✔
207	lastReconcileTimestamp.Gauge.SetToCurrentTime()	12✔
208		12✔
209	c.runAtMutex.Lock()	12✔
210	c.lastRunAt = time.Now()	12✔
211	c.runAtMutex.Unlock()	12✔
212		12✔
213	records, err := c.Registry.Records(ctx)	12✔
214	if err != nil {	12✔
NEW 215	registryErrorsTotal.Counter.Inc()	×
NEW 216	deprecatedRegistryErrors.Counter.Inc()	×
217	return err	×
218	}	×
219
220	registryEndpointsTotal.Gauge.Set(float64(len(records)))	12✔
221	regARecords, regAAAARecords := countAddressRecords(records)	12✔
222	registryARecords.Gauge.Set(float64(regARecords))	12✔
223	registryAAAARecords.Gauge.Set(float64(regAAAARecords))	12✔
224	ctx = context.WithValue(ctx, provider.RecordsContextKey, records)	12✔
225		12✔
226	endpoints, err := c.Source.Endpoints(ctx)	12✔
227	if err != nil {	12✔
NEW 228	sourceErrorsTotal.Counter.Inc()	×
NEW 229	deprecatedSourceErrors.Counter.Inc()	×
230	return err	×
231	}	×
232	sourceEndpointsTotal.Gauge.Set(float64(len(endpoints)))	12✔
233	srcARecords, srcAAAARecords := countAddressRecords(endpoints)	12✔
234	sourceARecords.Gauge.Set(float64(srcARecords))	12✔
235	sourceAAAARecords.Gauge.Set(float64(srcAAAARecords))	12✔
236	vARecords, vAAAARecords := countMatchingAddressRecords(endpoints, records)	12✔
237	verifiedARecords.Gauge.Set(float64(vARecords))	12✔
238	verifiedAAAARecords.Gauge.Set(float64(vAAAARecords))	12✔
239	endpoints, err = c.Registry.AdjustEndpoints(endpoints)	12✔
240	if err != nil {	12✔
241	return fmt.Errorf("adjusting endpoints: %w", err)	×
242	}	×
243	registryFilter := c.Registry.GetDomainFilter()	12✔
244		12✔
245	plan := &plan.Plan{	12✔
246	Policies: []plan.Policy{c.Policy},	12✔
247	Current: records,	12✔
248	Desired: endpoints,	12✔
249	DomainFilter: endpoint.MatchAllDomainFilters{c.DomainFilter, registryFilter},	12✔
250	ManagedRecords: c.ManagedRecordTypes,	12✔
251	ExcludeRecords: c.ExcludeRecordTypes,	12✔
252	OwnerID: c.Registry.OwnerID(),	12✔
253	}	12✔
254		12✔
255	plan = plan.Calculate()	12✔
256		12✔
257	if plan.Changes.HasChanges() {	21✔
258	err = c.Registry.ApplyChanges(ctx, plan.Changes)	9✔
259	if err != nil {	9✔
NEW 260	registryErrorsTotal.Counter.Inc()	×
NEW 261	deprecatedRegistryErrors.Counter.Inc()	×
262	return err	×
263	}	×
264	} else {	3✔
265	controllerNoChangesTotal.Counter.Inc()	3✔
266	log.Info("All records are already up to date")	3✔
267	}	3✔
268
269	lastSyncTimestamp.Gauge.SetToCurrentTime()	12✔
270		12✔
271	return nil	12✔
272	}
273
274	func earliest(r time.Time, times ...time.Time) time.Time {	4✔
275	for _, t := range times {	8✔
276	if t.Before(r) {	4✔
277	r = t	×
278	}	×
279	}
280	return r	4✔
281	}
282
283	func latest(r time.Time, times ...time.Time) time.Time {	4✔
284	for _, t := range times {	8✔
285	if t.After(r) {	4✔
286	r = t	×
287	}	×
288	}
289	return r	4✔
290	}
291
292	// Counts the intersections of A and AAAA records in endpoint and registry.
293	func countMatchingAddressRecords(endpoints []endpoint.Endpoint, registryRecords []endpoint.Endpoint) (int, int) {	12✔
294	recordsMap := make(map[string]map[string]struct{})	12✔
295	for _, regRecord := range registryRecords {	39✔
296	if _, found := recordsMap[regRecord.DNSName]; !found {	54✔
297	recordsMap[regRecord.DNSName] = make(map[string]struct{})	27✔
298	}	27✔
299	recordsMap[regRecord.DNSName][regRecord.RecordType] = struct{}{}	27✔
300	}
301	aCount := 0	12✔
302	aaaaCount := 0	12✔
303	for _, sourceRecord := range endpoints {	47✔
304	if _, found := recordsMap[sourceRecord.DNSName]; found {	56✔
305	if _, found := recordsMap[sourceRecord.DNSName][sourceRecord.RecordType]; found {	42✔
306	switch sourceRecord.RecordType {	21✔
307	case endpoint.RecordTypeA:	11✔
308	aCount++	11✔
309	case endpoint.RecordTypeAAAA:	8✔
310	aaaaCount++	8✔
311	}
312	}
313	}
314	}
315	return aCount, aaaaCount	12✔
316	}
317
318	func countAddressRecords(endpoints []*endpoint.Endpoint) (int, int) {	24✔
319	aCount := 0	24✔
320	aaaaCount := 0	24✔
321	for _, endPoint := range endpoints {	86✔
322	switch endPoint.RecordType {	62✔
323	case endpoint.RecordTypeA:	34✔
324	aCount++	34✔
325	case endpoint.RecordTypeAAAA:	24✔
326	aaaaCount++	24✔
327	}
328	}
329	return aCount, aaaaCount	24✔
330	}
331
332	// ScheduleRunOnce makes sure execution happens at most once per interval.
333	func (c *Controller) ScheduleRunOnce(now time.Time) {	4✔
334	c.runAtMutex.Lock()	4✔
335	defer c.runAtMutex.Unlock()	4✔
336	c.nextRunAt = latest(	4✔
337	c.lastRunAt.Add(c.MinEventSyncInterval),	4✔
338	earliest(	4✔
339	now.Add(5*time.Second),	4✔
340	c.nextRunAt,	4✔
341	),	4✔
342	)	4✔
343	}	4✔
344
345	func (c *Controller) ShouldRunOnce(now time.Time) bool {	13✔
346	c.runAtMutex.Lock()	13✔
347	defer c.runAtMutex.Unlock()	13✔
348	if now.Before(c.nextRunAt) {	20✔
349	return false	7✔
350	}	7✔
351	c.nextRunAt = now.Add(c.Interval)	6✔
352	return true	6✔
353	}
354
355	// Run runs RunOnce in a loop with a delay until context is canceled
356	func (c *Controller) Run(ctx context.Context) {	1✔
357	ticker := time.NewTicker(time.Second)	1✔
358	defer ticker.Stop()	1✔
359	for {	3✔
360	if c.ShouldRunOnce(time.Now()) {	4✔
361	if err := c.RunOnce(ctx); err != nil {	2✔
362	if errors.Is(err, provider.SoftError) {	×
363	log.Errorf("Failed to do run once: %v", err)	×
364	} else {	×
365	log.Fatalf("Failed to do run once: %v", err)	×
366	}	×
367	}
368	}
369	select {	2✔
370	case <-ticker.C:	1✔
371	case <-ctx.Done():	1✔
372	log.Info("Terminating main controller loop")	1✔
373	return	1✔
374	}
375	}
376	}

kubernetes-sigs / external-dns / 13654368756

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