xds/internal/di/k8sxds.go at f0fb6d797594d7e1a640988712b6c32fd0c143e5 · buildbuddy-io/xds · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
package di

import (
	"context"

	loadreportingservice "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v3"
	"github.com/envoyproxy/go-control-plane/pkg/server/v3"
	"github.com/google/wire"
	"github.com/wongnai/xds/debug"
	"github.com/wongnai/xds/meter"
	"github.com/wongnai/xds/report"
	"github.com/wongnai/xds/snapshot"
	"k8s.io/client-go/kubernetes"
	"k8s.io/klog/v2"
)

var K8sXdsSet = wire.NewSet(
	ProvideSnapshotter,
	ProvideXdsServer,
	ProvideXdsLogger,
	ProvideDebugServer,
	ProvideLRSServer,
)

func ProvideSnapshotter(ctx context.Context, k8sClient kubernetes.Interface, subZoneLabel snapshot.SubZoneLabel) (*snapshot.Snapshotter, func()) {
	stopCtx, stop := context.WithCancel(ctx)
	snapshotter := snapshot.New(k8sClient, subZoneLabel)

	go func() {
		err := snapshotter.Start(stopCtx)
		if err != nil {
			klog.Fatal(err)
		}
	}()

	return snapshotter, stop
}

func ProvideXdsServer(ctx context.Context, snapshotter *snapshot.Snapshotter, logger server.CallbackFuncs) (server.Server, func()) {
	stopCtx, stop := context.WithCancel(ctx)

	return server.NewServer(stopCtx, snapshotter.MuxCache(), logger), stop
}

func ProvideXdsLogger() server.CallbackFuncs {
	return meter.NewXdsServerCallbackFuncs()
}

// ProvideDebugServer create a debug server and immediately starts it
func ProvideDebugServer(snapshotter *snapshot.Snapshotter) *debug.Server {
	server := debug.New(snapshotter.MuxCache())

	go server.ListenAndServe()

	return server
}

type StatsIntervalSeconds = int64

func ProvideLRSServer(statsIntervalSeconds StatsIntervalSeconds) loadreportingservice.LoadReportingServiceServer {
	return report.NewServer(report.WithStatsIntervalInSeconds(statsIntervalSeconds))
}