kuma

1
package hds
2

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

7
	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
8
	envoy_service_health "github.com/envoyproxy/go-control-plane/envoy/service/health/v3"
9

10
	config_core "github.com/kumahq/kuma/pkg/config/core"
11
	"github.com/kumahq/kuma/pkg/core"
12
	core_runtime "github.com/kumahq/kuma/pkg/core/runtime"
13
	"github.com/kumahq/kuma/pkg/core/user"
14
	"github.com/kumahq/kuma/pkg/hds/authn"
15
	hds_callbacks "github.com/kumahq/kuma/pkg/hds/callbacks"
16
	hds_metrics "github.com/kumahq/kuma/pkg/hds/metrics"
17
	hds_server "github.com/kumahq/kuma/pkg/hds/server"
18
	"github.com/kumahq/kuma/pkg/hds/tracker"
19
	util_xds "github.com/kumahq/kuma/pkg/util/xds"
20
	util_xds_v3 "github.com/kumahq/kuma/pkg/util/xds/v3"
21
)
22

23
var hdsServerLog = core.Log.WithName("hds-server")
24

25
func Setup(rt core_runtime.Runtime) error {
26
	if rt.Config().Mode == config_core.Global {
27
		return nil
28
	}
29
	if !rt.Config().DpServer.Hds.Enabled {
30
		return nil
31
	}
32

33
	snapshotCache := util_xds_v3.NewSnapshotCache(false, hasher{}, util_xds.NewLogger(hdsServerLog))
34

35
	callbacks, err := DefaultCallbacks(rt, snapshotCache)
36
	if err != nil {
37
		return err
38
	}
39

40
	srv := hds_server.New(user.Ctx(context.Background(), user.ControlPlane), snapshotCache, callbacks)
41

42
	hdsServerLog.Info("registering Health Discovery Service in Dataplane Server")
43
	envoy_service_health.RegisterHealthDiscoveryServiceServer(rt.DpServer().GrpcServer(), srv)
44
	return nil
45
}
46

47
func DefaultCallbacks(rt core_runtime.Runtime, cache util_xds_v3.SnapshotCache) (hds_callbacks.Callbacks, error) {
48
	metrics, err := hds_metrics.NewMetrics(rt.Metrics())
49
	if err != nil {
50
		return nil, err
51
	}
52

53
	return hds_callbacks.Chain{
54
		authn.NewCallbacks(rt.ResourceManager(), rt.XDS().DpProxyAuthenticator, authn.DPNotFoundRetry{
55
			// Usually the difference between DP is created from ADS and HDS is initiated is less than 1 second, but just in case we set this higher.
56
			Backoff:  1 * time.Second,
57
			MaxTimes: 30,
58
		}),
59
		tracker.NewCallbacks(
60
			hdsServerLog,
61
			rt.ResourceManager(),
62
			rt.ReadOnlyResourceManager(),
63
			cache,
64
			rt.Config().DpServer.Hds,
65
			hasher{},
66
			metrics,
67
			rt.Config().GetEnvoyAdminPort(),
68
		),
69
	}, nil
70
}
71

72
type hasher struct{}
73

74
func (_ hasher) ID(node *envoy_core.Node) string {
75
	return node.Id
76
}
77