kuma

1
package reconcile
2

3
import (
4
	"context"
5

6
	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
7
	envoy_cache "github.com/envoyproxy/go-control-plane/pkg/cache/v3"
8

9
	config_core "github.com/kumahq/kuma/pkg/config/core"
10
	"github.com/kumahq/kuma/pkg/core"
11
	"github.com/kumahq/kuma/pkg/util/xds"
12
	util_xds_v3 "github.com/kumahq/kuma/pkg/util/xds/v3"
13
)
14

15
var log = core.Log.WithName("kds").WithName("reconcile")
16

17
func NewReconciler(
18
	hasher envoy_cache.NodeHash,
19
	cache util_xds_v3.SnapshotCache,
20
	generator SnapshotGenerator,
21
	versioner util_xds_v3.SnapshotVersioner,
22
	mode config_core.CpMode,
23
	statsCallbacks xds.StatsCallbacks,
24
) Reconciler {
25
	return &reconciler{
26
		hasher:         hasher,
27
		cache:          cache,
28
		generator:      generator,
29
		versioner:      versioner,
30
		mode:           mode,
31
		statsCallbacks: statsCallbacks,
32
	}
33
}
34

35
type reconciler struct {
36
	hasher         envoy_cache.NodeHash
37
	cache          util_xds_v3.SnapshotCache
38
	generator      SnapshotGenerator
39
	versioner      util_xds_v3.SnapshotVersioner
40
	mode           config_core.CpMode
41
	statsCallbacks xds.StatsCallbacks
42
}
43

44
func (r *reconciler) Clear(ctx context.Context, node *envoy_core.Node) {
45
	id := r.hasher.ID(node)
46
	snapshot := r.cache.ClearSnapshot(id)
47
	if snapshot == nil {
48
		return
49
	}
50
	for _, typ := range snapshot.GetSupportedTypes() {
51
		r.statsCallbacks.DiscardConfig(snapshot.GetVersion(typ))
52
	}
53
}
54

55
func (r *reconciler) Reconcile(ctx context.Context, node *envoy_core.Node) error {
56
	new, err := r.generator.GenerateSnapshot(ctx, node)
57
	if err != nil {
58
		return err
59
	}
60
	if err := new.Consistent(); err != nil {
61
		return err
62
	}
63
	id := r.hasher.ID(node)
64
	old, _ := r.cache.GetSnapshot(id)
65
	new = r.versioner.Version(new, old)
66
	r.logChanges(new, old, node)
67
	r.meterConfigReadyForDelivery(new, old)
68
	return r.cache.SetSnapshot(id, new)
69
}
70

71
func (r *reconciler) logChanges(new util_xds_v3.Snapshot, old util_xds_v3.Snapshot, node *envoy_core.Node) {
72
	for _, typ := range new.GetSupportedTypes() {
73
		if old != nil && old.GetVersion(typ) != new.GetVersion(typ) {
74
			client := node.Id
75
			if r.mode == config_core.Zone {
76
				// we need to override client name because Zone is always a client to Global (on gRPC level)
77
				client = "global"
78
			}
79
			log.Info("detected changes in the resources. Sending changes to the client.", "resourceType", typ, "client", client)
80
		}
81
	}
82
}
83

84
func (r *reconciler) meterConfigReadyForDelivery(new util_xds_v3.Snapshot, old util_xds_v3.Snapshot) {
85
	for _, typ := range new.GetSupportedTypes() {
86
		if old == nil || old.GetVersion(typ) != new.GetVersion(typ) {
87
			r.statsCallbacks.ConfigReadyForDelivery(new.GetVersion(typ))
88
		}
89
	}
90
}
91