-
Notifications
You must be signed in to change notification settings - Fork 9
/
cluster.go
117 lines (93 loc) · 2.93 KB
/
cluster.go
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
// Copyright 2022 Namespace Labs Inc; All rights reserved.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
package kubernetes
import (
"context"
"sync"
k8s "k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"namespacelabs.dev/foundation/framework/kubernetes/kubedef"
"namespacelabs.dev/foundation/internal/runtime"
"namespacelabs.dev/foundation/internal/runtime/kubernetes/client"
"namespacelabs.dev/foundation/internal/tcache"
"namespacelabs.dev/foundation/std/cfg"
"namespacelabs.dev/foundation/std/tasks"
)
type Cluster struct {
cli *k8s.Clientset
computedClient client.Prepared
config cfg.Configuration
FetchSystemInfo func(context.Context) (*kubedef.SystemInfo, error)
ClusterAttachedState
}
type ClusterAttachedState struct {
mu sync.Mutex
attachedState map[string]*state
}
type state struct {
mu sync.Mutex
resolved bool
value any
err error
}
var _ kubedef.KubeCluster = &Cluster{}
func ConnectToCluster(ctx context.Context, config cfg.Configuration) (*Cluster, error) {
cli, err := client.NewClient(ctx, config)
if err != nil {
return nil, err
}
deferredSystemInfo := tcache.NewDeferred(tasks.Action("kubernetes.fetch-system-info"), func(ctx context.Context) (*kubedef.SystemInfo, error) {
return computeSystemInfo(ctx, cli.Clientset)
})
return &Cluster{
cli: cli.Clientset,
computedClient: *cli,
config: config,
FetchSystemInfo: deferredSystemInfo.Get,
}, nil
}
func (u *Cluster) Class() runtime.Class {
return kubernetesClass{}
}
func (u *Cluster) Planner(env cfg.Context) runtime.Planner {
return NewPlanner(env, u.SystemInfo)
}
func (u *Cluster) RESTConfig() *rest.Config {
return u.computedClient.RESTConfig
}
func (u *Cluster) PreparedClient() client.Prepared {
return u.computedClient
}
func (u *Cluster) Bind(env cfg.Context) (runtime.ClusterNamespace, error) {
return &ClusterNamespace{cluster: u, target: newTarget(env)}, nil
}
func (r *Cluster) EnsureState(ctx context.Context, key string) (any, error) {
return r.ClusterAttachedState.EnsureState(ctx, key, r.config, r, nil)
}
func (r *ClusterAttachedState) EnsureState(ctx context.Context, stateKey string, config cfg.Configuration, cluster runtime.Cluster, key *string) (any, error) {
r.mu.Lock()
if r.attachedState == nil {
r.attachedState = map[string]*state{}
}
computedKey := stateKey
if key != nil {
computedKey += ":" + *key
}
if r.attachedState[computedKey] == nil {
r.attachedState[computedKey] = &state{}
}
state := r.attachedState[computedKey]
r.mu.Unlock()
state.mu.Lock()
defer state.mu.Unlock()
if !state.resolved {
if key != nil {
state.value, state.err = runtime.PrepareKeyed(ctx, stateKey, config, cluster, *key)
} else {
state.value, state.err = runtime.Prepare(ctx, stateKey, config, cluster)
}
state.resolved = true
}
return state.value, state.err
}