kubelatte-ce

1
package kubeapi
2

3
import (
4
	"context"
5
	"fmt"
6
	"gitverse.ru/synapse/kubelatte/pkg/api/v1alpha1"
7
	"gitverse.ru/synapse/kubelatte/pkg/observability/logger"
8
	"gitverse.ru/synapse/kubelatte/pkg/operator/controllers/clientset"
9
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
10
	"k8s.io/apimachinery/pkg/runtime"
11
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
12
	"k8s.io/client-go/kubernetes"
13
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
14
	"k8s.io/client-go/rest"
15
	ctrl "sigs.k8s.io/controller-runtime"
16
	"strings"
17
	"sync"
18
)
19

20
type ResourceMeta struct {
21
	Name         string
22
	Plural       string
23
	Path         string
24
	Namespaced   bool
25
	GroupVersion string
26
}
27

28
type syncClient struct {
29
	client *clientset.V1Alpha1Client
30
	mutex  sync.Mutex
31
}
32

33
var (
34
	scheme = runtime.NewScheme()
35
	client = &syncClient{mutex: sync.Mutex{}}
36
)
37

38
func getResourceFromKubeApi(ctx context.Context, client rest.Interface, resourceMeta *ResourceMeta, resourceName string, namespaceName string) ([]byte, error) {
39
	request := client.Get().
40
		Resource(resourceMeta.Plural).
41
		Name(resourceName).
42
		Prefix(resourceMeta.Path, resourceMeta.GroupVersion)
43

44
	if resourceMeta.Namespaced {
45
		request.Namespace(namespaceName)
46
	}
47

48
	res, err := request.Do(ctx).Raw()
49

50
	return res, err
51
}
52

53
var GetResource = func(ctx context.Context, resourceMeta *ResourceMeta, resourceName string, namespace string) ([]byte, error) {
54
	log := logger.FromContext(ctx)
55
	cliKBLT, err := GetKBLTClient()
56
	if err != nil {
57
		log.Errorf("Failed getting KBLT kube client")
58
		return nil, err
59
	}
60

61
	res, err := getResourceFromKubeApi(ctx, cliKBLT.RestClient, resourceMeta, resourceName, namespace)
62
	if err != nil {
63
		cliKUBE := GetClient()
64
		res, err = getResourceFromKubeApi(ctx, cliKUBE.RESTClient(), resourceMeta, resourceName, namespace)
65
	}
66

67
	return res, err
68
}
69

70
var GetKBLTClient = func() (*clientset.V1Alpha1Client, error) {
71
	client.mutex.Lock()
72
	defer client.mutex.Unlock()
73

74
	log := logger.FromContext(context.Background())
75
	if client.client == nil {
76
		var err error
77
		utilruntime.Must(clientgoscheme.AddToScheme(scheme))
78
		utilruntime.Must(v1alpha1.AddToScheme(scheme))
79
		//+kubebuilder:scaffold:scheme
80

81
		configConfs := ctrl.GetConfigOrDie()
82
		err = v1alpha1.AddToScheme(clientgoscheme.Scheme)
83
		if err != nil {
84
			log.Errorf("v1alpha1.AddToScheme error: %s", err.Error())
85
			return nil, err
86
		}
87
		client.client, err = clientset.NewForConfig(configConfs)
88
		if err != nil {
89
			return nil, err
90
		}
91
	}
92

93
	return client.client, nil
94
}
95

96
var GetClient = func() *kubernetes.Clientset {
97
	log := logger.FromContext(context.Background())
98
	config := ctrl.GetConfigOrDie()
99
	config.QPS = 100
100
	config.Burst = 50
101
	clt, err := kubernetes.NewForConfig(config)
102
	if err != nil {
103
		log.Errorf("GetCoreClient: failed to init kubernetes client: %s", err.Error())
104
		panic(err.Error())
105
	}
106

107
	return clt
108
}
109

110
var _resourceMetaCache = map[string]ResourceMeta{}
111
var _resourcesCache []*v1.APIResourceList
112

113
var InvalidateCacheAndTryGetResourceMeta = func(client *kubernetes.Clientset, kind string, version string) *ResourceMeta {
114
	_resourcesCache = []*v1.APIResourceList{}
115
	return GetResourceMeta(client, kind, version)
116
}
117

118
var GetResourceMeta = func(client *kubernetes.Clientset, kind string, version string) *ResourceMeta {
119
	meta := CheckResourceMeta(client, kind, version)
120
	return meta
121
}
122

123
var CheckResourceMeta = func(client *kubernetes.Clientset, kind string, version string) *ResourceMeta {
124
	log := logger.FromContext(context.Background())
125
	name := fmt.Sprintf("%s-%s", kind, version)
126
	result, ok := _resourceMetaCache[name]
127
	if ok {
128
		return &result
129
	}
130

131
	if len(_resourcesCache) == 0 {
132
		var err error
133
		_, _resourcesCache, err = client.ServerGroupsAndResources()
134
		if err != nil {
135
			log.Errorf("FactoryController: GetResourceMeta failed: %s", err.Error())
136
			return nil
137
		}
138
	}
139

140
	path := ""
141
	for _, resource := range _resourcesCache {
142
		if strings.ToLower(resource.GroupVersion) == strings.ToLower(version) {
143
			for _, apiResource := range resource.APIResources {
144
				if strings.ToLower(apiResource.Kind) == strings.ToLower(kind) {
145
					if strings.ToLower(resource.GroupVersion) == "v1" {
146
						path = "api"
147
					} else {
148
						path = "apis"
149
					}
150

151
					meta := ResourceMeta{
152
						Name:         apiResource.Kind,
153
						Plural:       apiResource.Name,
154
						Path:         path,
155
						Namespaced:   apiResource.Namespaced,
156
						GroupVersion: resource.GroupVersion,
157
					}
158
					_resourceMetaCache[name] = meta
159

160
					return &meta
161
				}
162
			}
163
		}
164
	}
165

166
	return nil
167
}
168

169
var GetRawResourceFromKubeApi = func(ctx context.Context, client rest.Interface, resourceMeta *ResourceMeta, resourceName string, namespaceName string) ([]byte, error) {
170
	request := client.Get().
171
		Resource(resourceMeta.Plural).
172
		Name(resourceName).
173
		Prefix(resourceMeta.Path, resourceMeta.GroupVersion)
174

175
	if resourceMeta.Namespaced {
176
		request.Namespace(namespaceName)
177
	}
178

179
	return request.DoRaw(ctx)
180
}
181

182
var CreateResourceInKubeApi = func(ctx context.Context, client rest.Interface, resourceMeta *ResourceMeta, jsBody []byte, namespaceName string) ([]byte, error, error) {
183
	request := client.Post().
184
		Resource(resourceMeta.Plural).
185
		Prefix(resourceMeta.Path, resourceMeta.GroupVersion).
186
		Body(jsBody)
187

188
	if resourceMeta.Namespaced {
189
		request.Namespace(namespaceName)
190
	}
191

192
	result := request.Do(ctx)
193
	e := result.Error()
194
	raw, err := result.Raw()
195
	return raw, err, e
196
}
197