kuma

1
package reconcile
2

3
import (
4
	"context"
5
	"strings"
6

7
	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
8
	envoy_types "github.com/envoyproxy/go-control-plane/pkg/cache/types"
9

10
	config_store "github.com/kumahq/kuma/pkg/config/core/resources/store"
11
	core_manager "github.com/kumahq/kuma/pkg/core/resources/manager"
12
	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
13
	"github.com/kumahq/kuma/pkg/core/resources/registry"
14
	"github.com/kumahq/kuma/pkg/kds"
15
	"github.com/kumahq/kuma/pkg/kds/cache"
16
	"github.com/kumahq/kuma/pkg/kds/util"
17
	util_xds_v3 "github.com/kumahq/kuma/pkg/util/xds/v3"
18
)
19

20
type (
21
	ResourceFilter func(ctx context.Context, clusterID string, features kds.Features, r core_model.Resource) bool
22
	ResourceMapper func(features kds.Features, r core_model.Resource) (core_model.Resource, error)
23
)
24

25
func NoopResourceMapper(_ kds.Features, r core_model.Resource) (core_model.Resource, error) {
26
	return r, nil
27
}
28

29
func Any(context.Context, string, kds.Features, core_model.Resource) bool {
30
	return true
31
}
32

33
func TypeIs(rtype core_model.ResourceType) func(core_model.Resource) bool {
34
	return func(r core_model.Resource) bool {
35
		return r.Descriptor().Name == rtype
36
	}
37
}
38

39
func IsKubernetes(storeType config_store.StoreType) func(core_model.Resource) bool {
40
	return func(_ core_model.Resource) bool {
41
		return storeType == config_store.KubernetesStore
42
	}
43
}
44

45
func ScopeIs(s core_model.ResourceScope) func(core_model.Resource) bool {
46
	return func(r core_model.Resource) bool {
47
		return r.Descriptor().Scope == s
48
	}
49
}
50

51
func NameHasPrefix(prefix string) func(core_model.Resource) bool {
52
	return func(r core_model.Resource) bool {
53
		return strings.HasPrefix(r.GetMeta().GetName(), prefix)
54
	}
55
}
56

57
func Not(f func(core_model.Resource) bool) func(core_model.Resource) bool {
58
	return func(r core_model.Resource) bool {
59
		return !f(r)
60
	}
61
}
62

63
func And(fs ...func(core_model.Resource) bool) func(core_model.Resource) bool {
64
	return func(r core_model.Resource) bool {
65
		for _, f := range fs {
66
			if !f(r) {
67
				return false
68
			}
69
		}
70
		return true
71
	}
72
}
73

74
func If(condition func(core_model.Resource) bool, m ResourceMapper) ResourceMapper {
75
	return func(features kds.Features, r core_model.Resource) (core_model.Resource, error) {
76
		if condition(r) {
77
			return m(features, r)
78
		}
79
		return r, nil
80
	}
81
}
82

83
func NewSnapshotGenerator(resourceManager core_manager.ReadOnlyResourceManager, types []core_model.ResourceType, filter ResourceFilter, mapper ResourceMapper) SnapshotGenerator {
84
	return &snapshotGenerator{
85
		resourceManager: resourceManager,
86
		resourceTypes:   types,
87
		resourceFilter:  filter,
88
		resourceMapper:  mapper,
89
	}
90
}
91

92
type snapshotGenerator struct {
93
	resourceManager core_manager.ReadOnlyResourceManager
94
	resourceTypes   []core_model.ResourceType
95
	resourceFilter  ResourceFilter
96
	resourceMapper  ResourceMapper
97
}
98

99
func (s *snapshotGenerator) GenerateSnapshot(ctx context.Context, node *envoy_core.Node) (util_xds_v3.Snapshot, error) {
100
	builder := cache.NewSnapshotBuilder()
101
	for _, typ := range s.resourceTypes {
102
		resources, err := s.getResources(ctx, typ, node)
103
		if err != nil {
104
			return nil, err
105
		}
106
		builder = builder.With(string(typ), resources)
107
	}
108

109
	return builder.Build(""), nil
110
}
111

112
func (s *snapshotGenerator) getResources(ctx context.Context, typ core_model.ResourceType, node *envoy_core.Node) ([]envoy_types.Resource, error) {
113
	rlist, err := registry.Global().NewList(typ)
114
	if err != nil {
115
		return nil, err
116
	}
117
	if err := s.resourceManager.List(ctx, rlist); err != nil {
118
		return nil, err
119
	}
120

121
	resources, err := s.mapper(s.filter(ctx, rlist, node), node)
122
	if err != nil {
123
		return nil, err
124
	}
125

126
	return util.ToEnvoyResources(resources)
127
}
128

129
func (s *snapshotGenerator) filter(ctx context.Context, rs core_model.ResourceList, node *envoy_core.Node) core_model.ResourceList {
130
	features := getFeatures(node)
131

132
	rv := registry.Global().MustNewList(rs.GetItemType())
133
	for _, r := range rs.GetItems() {
134
		if s.resourceFilter(ctx, node.GetId(), features, r) {
135
			_ = rv.AddItem(r)
136
		}
137
	}
138
	return rv
139
}
140

141
func (s *snapshotGenerator) mapper(rs core_model.ResourceList, node *envoy_core.Node) (core_model.ResourceList, error) {
142
	features := getFeatures(node)
143

144
	rv := registry.Global().MustNewList(rs.GetItemType())
145
	for _, r := range rs.GetItems() {
146
		resource, err := s.resourceMapper(features, r)
147
		if err != nil {
148
			return nil, err
149
		}
150

151
		if err := rv.AddItem(resource); err != nil {
152
			return nil, err
153
		}
154
	}
155

156
	return rv, nil
157
}
158

159
func getFeatures(node *envoy_core.Node) kds.Features {
160
	features := kds.Features{}
161
	for _, value := range node.GetMetadata().GetFields()[kds.MetadataFeatures].GetListValue().GetValues() {
162
		features[value.GetStringValue()] = true
163
	}
164
	return features
165
}
166