Skip to content

Commit bab7efe

Browse files
bchobcho
committed
Add preflight command
1 parent 52ff2a0 commit bab7efe

3 files changed

Lines changed: 381 additions & 0 deletions

File tree

‎cmd/aks-flex-node/main.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/spf13/cobra"
1212

1313
"github.com/Azure/AKSFlexNode/pkg/cmd/daemon"
14+
"github.com/Azure/AKSFlexNode/pkg/cmd/preflight"
1415
"github.com/Azure/AKSFlexNode/pkg/cmd/reset"
1516
"github.com/Azure/AKSFlexNode/pkg/cmd/start"
1617
"github.com/Azure/AKSFlexNode/pkg/cmd/token"
@@ -25,6 +26,7 @@ func main() {
2526
}
2627

2728
rootCmd.AddCommand(start.NewCommand())
29+
rootCmd.AddCommand(preflight.NewCommand())
2830
rootCmd.AddCommand(daemon.NewCommand())
2931
rootCmd.AddCommand(reset.NewCommand())
3032
rootCmd.AddCommand(version.NewCommand())

‎pkg/cmd/preflight/preflight.go‎

Lines changed: 273 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,273 @@
1+
package preflight
2+
3+
import (
4+
"context"
5+
"crypto/tls"
6+
"crypto/x509"
7+
"encoding/base64"
8+
"encoding/json"
9+
"fmt"
10+
"io"
11+
"log/slog"
12+
"net/http"
13+
"net/url"
14+
"os"
15+
"strings"
16+
"time"
17+
18+
"github.com/spf13/cobra"
19+
20+
"github.com/Azure/AKSFlexNode/pkg/config"
21+
"github.com/Azure/AKSFlexNode/pkg/logger"
22+
agentconfig "github.com/Azure/unbounded/pkg/agent/config"
23+
"github.com/Azure/unbounded/pkg/agent/goalstates"
24+
"github.com/Azure/unbounded/pkg/agent/phases/host"
25+
"github.com/Azure/unbounded/pkg/agent/phases/rootfs"
26+
agentpreflight "github.com/Azure/unbounded/pkg/agent/preflight"
27+
)
28+
29+
const checkAPIServerReachableName = "api-server-reachable"
30+
31+
type handler struct {
32+
configPath string
33+
ignorePreflightErrors []string
34+
failOnWarnings bool
35+
output string
36+
writer io.Writer
37+
}
38+
39+
// NewCommand returns the preflight command.
40+
func NewCommand() *cobra.Command {
41+
h := &handler{writer: os.Stdout}
42+
43+
cmd := &cobra.Command{
44+
Use: "preflight",
45+
Short: "Run non-mutating preflight checks",
46+
Long: "Run non-mutating preflight checks for the host and AKS Flex Node configuration before node bootstrap.",
47+
RunE: func(cmd *cobra.Command, args []string) error {
48+
return h.execute(cmd.Context())
49+
},
50+
}
51+
52+
cmd.Flags().StringVar(&h.configPath, "config", "", "Path to configuration JSON file (required)")
53+
_ = cmd.MarkFlagRequired("config")
54+
cmd.Flags().StringSliceVar(
55+
&h.ignorePreflightErrors,
56+
"ignore-preflight-errors",
57+
nil,
58+
"Comma-separated preflight check names whose errors should be reported as warnings",
59+
)
60+
cmd.Flags().BoolVar(&h.failOnWarnings, "fail-on-warnings", false, "Fail when any preflight warning is returned")
61+
cmd.Flags().StringVar(&h.output, "output", "text", "Output format: text or json")
62+
63+
return cmd
64+
}
65+
66+
func (h *handler) execute(ctx context.Context) error {
67+
cfg, err := config.LoadConfig(h.configPath)
68+
if err != nil {
69+
return fmt.Errorf("failed to load config from %s: %w", h.configPath, err)
70+
}
71+
log := logger.CreateLogger(cfg.Agent.LogLevel, cfg.Agent.LogDir)
72+
73+
agentCfg := config.ToAgentConfig(cfg, goalstates.NSpawnMachineKube1)
74+
gs, err := goalstates.ResolveMachine(log, agentCfg, goalstates.NSpawnMachineKube1, nil)
75+
if err != nil {
76+
return fmt.Errorf("preflight failed to resolve goal state: %w", err)
77+
}
78+
79+
checks := agentpreflight.Flatten(
80+
hostChecks(log),
81+
[]agentpreflight.Checker{checkAPIServerReachable(log, agentCfg)},
82+
rootFSChecks(log, gs),
83+
)
84+
85+
report := agentpreflight.Run(ctx, checks, agentpreflight.Options{
86+
IgnoreErrors: h.ignorePreflightErrors,
87+
FailOnWarnings: h.failOnWarnings,
88+
})
89+
90+
switch strings.ToLower(h.output) {
91+
case "", "text":
92+
if err := writeText(h.writer, report); err != nil {
93+
return err
94+
}
95+
case "json":
96+
enc := json.NewEncoder(h.writer)
97+
enc.SetIndent("", " ")
98+
if err := enc.Encode(report); err != nil {
99+
return err
100+
}
101+
default:
102+
return fmt.Errorf("unsupported output format %q", h.output)
103+
}
104+
105+
return report.Err(h.failOnWarnings)
106+
}
107+
108+
func hostChecks(log *slog.Logger) []agentpreflight.Checker {
109+
return []agentpreflight.Checker{
110+
host.CheckIsPrivilegedUser(log),
111+
host.CheckHostPackages(log),
112+
host.CheckHostOSConfiguration(log),
113+
host.CheckNSpawnRuntime(log),
114+
host.CheckDockerActive(log),
115+
host.CheckSwapActive(log),
116+
host.CheckDiskSpace(log),
117+
host.CheckCgroups(log),
118+
host.CheckNvidiaDriver(log),
119+
}
120+
}
121+
122+
func rootFSChecks(log *slog.Logger, gs *goalstates.MachineGoalState) []agentpreflight.Checker {
123+
var rootFS *goalstates.RootFS
124+
if gs != nil {
125+
rootFS = gs.RootFS
126+
}
127+
128+
return []agentpreflight.Checker{
129+
rootfs.CheckOCIImageReachable(log, rootFS),
130+
rootfs.CheckKubernetesArtifacts(log, rootFS),
131+
rootfs.CheckCRIArtifacts(log, rootFS),
132+
rootfs.CheckCNIArtifacts(log, rootFS),
133+
rootfs.CheckNSpawnMachineProvisioning(log, rootFS),
134+
}
135+
}
136+
137+
type apiServerReachableChecker struct {
138+
log *slog.Logger
139+
config *agentconfig.AgentConfig
140+
httpClient *http.Client
141+
}
142+
143+
func checkAPIServerReachable(log *slog.Logger, cfg *agentconfig.AgentConfig) agentpreflight.Checker {
144+
return apiServerReachableChecker{log: log, config: cfg}
145+
}
146+
147+
func (c apiServerReachableChecker) Name() string { return checkAPIServerReachableName }
148+
149+
func (c apiServerReachableChecker) Check(ctx context.Context) []agentpreflight.Result {
150+
if c.config == nil {
151+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "agent config is missing")
152+
}
153+
154+
if errs := c.validateClusterCredentials(); len(errs) > 0 {
155+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster credentials", "%s", strings.Join(errs, "; "))
156+
}
157+
158+
apiServer := c.config.Kubelet.ApiServer
159+
if strings.TrimSpace(apiServer) == "" {
160+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "API server is required")
161+
}
162+
163+
parsed, err := url.Parse(apiServer)
164+
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
165+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "API server endpoint is invalid")
166+
}
167+
168+
client := c.httpClient
169+
if client == nil {
170+
client = c.httpClientWithCA()
171+
}
172+
173+
req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(apiServer, "/")+"/readyz", http.NoBody)
174+
if err != nil {
175+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "API server request could not be created")
176+
}
177+
178+
resp, err := client.Do(req)
179+
if err != nil {
180+
c.log.Debug("preflight API server probe failed")
181+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "API server is not reachable")
182+
}
183+
defer resp.Body.Close() //nolint:errcheck // best effort close
184+
185+
if resp.StatusCode >= http.StatusInternalServerError {
186+
return agentpreflight.ResultsError(checkAPIServerReachableName, "cluster API server", "API server returned status %d", resp.StatusCode)
187+
}
188+
189+
return agentpreflight.ResultsOK(checkAPIServerReachableName, "cluster API server", "API server is reachable")
190+
}
191+
192+
func (c apiServerReachableChecker) validateClusterCredentials() []string {
193+
var errs []string
194+
if _, err := base64.StdEncoding.DecodeString(c.config.Cluster.CaCertBase64); err != nil {
195+
errs = append(errs, "cluster CA data is invalid")
196+
}
197+
198+
if err := c.config.Kubelet.Auth.Validate(); err != nil {
199+
errs = append(errs, "bootstrap credential is invalid")
200+
}
201+
202+
return errs
203+
}
204+
205+
func (c apiServerReachableChecker) httpClientWithCA() *http.Client {
206+
transport := &http.Transport{}
207+
if defaultTransport, ok := http.DefaultTransport.(*http.Transport); ok {
208+
transport = defaultTransport.Clone()
209+
}
210+
211+
caCertData, err := base64.StdEncoding.DecodeString(c.config.Cluster.CaCertBase64)
212+
if err == nil && len(caCertData) > 0 {
213+
pool, err := x509.SystemCertPool()
214+
if err != nil {
215+
pool = x509.NewCertPool()
216+
}
217+
pool.AppendCertsFromPEM(caCertData)
218+
transport.TLSClientConfig = &tls.Config{RootCAs: pool} //nolint:gosec // uses configured root CAs.
219+
}
220+
221+
return &http.Client{Timeout: 10 * time.Second, Transport: transport}
222+
}
223+
224+
func writeText(w io.Writer, report agentpreflight.Report) error {
225+
if _, err := fmt.Fprintln(w, "[preflight] Running AKS Flex Node preflight checks"); err != nil {
226+
return err
227+
}
228+
229+
var errors []agentpreflight.Result
230+
for _, result := range report.Checks {
231+
switch result.Severity {
232+
case agentpreflight.SeverityOK:
233+
if err := writeResult(w, "OK", result); err != nil {
234+
return err
235+
}
236+
case agentpreflight.SeverityError:
237+
errors = append(errors, result)
238+
case agentpreflight.SeverityWarning:
239+
if err := writeResult(w, "WARNING", result); err != nil {
240+
return err
241+
}
242+
}
243+
}
244+
245+
if len(errors) == 0 {
246+
return nil
247+
}
248+
249+
if _, err := fmt.Fprintln(w, "[preflight] Some fatal errors occurred:"); err != nil {
250+
return err
251+
}
252+
for _, result := range errors {
253+
if err := writeResult(w, "ERROR", result); err != nil {
254+
return err
255+
}
256+
}
257+
258+
_, err := fmt.Fprintln(w, "[preflight] If you know what you are doing, you can make a check non-fatal with `--ignore-preflight-errors=...`")
259+
return err
260+
}
261+
262+
func writeResult(w io.Writer, status string, result agentpreflight.Result) error {
263+
if _, err := fmt.Fprintf(w, "\t[%s %s]: %s", status, result.Name, result.Message); err != nil {
264+
return err
265+
}
266+
if result.Target != "" {
267+
if _, err := fmt.Fprintf(w, " (target: %s)", result.Target); err != nil {
268+
return err
269+
}
270+
}
271+
_, err := fmt.Fprintln(w)
272+
return err
273+
}

‎pkg/cmd/preflight/preflight_test.go‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
package preflight
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"log/slog"
7+
"net/http"
8+
"net/http/httptest"
9+
"strings"
10+
"testing"
11+
12+
agentconfig "github.com/Azure/unbounded/pkg/agent/config"
13+
agentpreflight "github.com/Azure/unbounded/pkg/agent/preflight"
14+
)
15+
16+
func TestNewCommand(t *testing.T) {
17+
t.Parallel()
18+
19+
cmd := NewCommand()
20+
if cmd.Use != "preflight" {
21+
t.Fatalf("Use = %q, want preflight", cmd.Use)
22+
}
23+
24+
for _, flag := range []string{"config", "ignore-preflight-errors", "fail-on-warnings", "output"} {
25+
if cmd.Flags().Lookup(flag) == nil {
26+
t.Fatalf("expected flag %q", flag)
27+
}
28+
}
29+
}
30+
31+
func TestWriteText(t *testing.T) {
32+
t.Parallel()
33+
34+
report := agentpreflight.Report{
35+
Checks: []agentpreflight.Result{
36+
agentpreflight.OK("ok-check", "ok target", "all good"),
37+
agentpreflight.Warning("warn-check", "warn target", "be careful"),
38+
agentpreflight.Error("error-check", "error target", "bad thing"),
39+
},
40+
}
41+
42+
var out bytes.Buffer
43+
if err := writeText(&out, report); err != nil {
44+
t.Fatalf("writeText() error = %v", err)
45+
}
46+
47+
got := out.String()
48+
for _, want := range []string{
49+
"[preflight] Running AKS Flex Node preflight checks",
50+
"[OK ok-check]: all good (target: ok target)",
51+
"[WARNING warn-check]: be careful (target: warn target)",
52+
"[ERROR error-check]: bad thing (target: error target)",
53+
} {
54+
if !strings.Contains(got, want) {
55+
t.Fatalf("writeText() output missing %q\n%s", want, got)
56+
}
57+
}
58+
}
59+
60+
func TestAPIServerReachableChecker(t *testing.T) {
61+
t.Parallel()
62+
63+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
64+
if r.URL.Path != "/readyz" {
65+
t.Fatalf("path = %q, want /readyz", r.URL.Path)
66+
}
67+
w.WriteHeader(http.StatusOK)
68+
}))
69+
t.Cleanup(srv.Close)
70+
71+
checker := checkAPIServerReachable(slog.New(slog.DiscardHandler), &agentconfig.AgentConfig{
72+
Cluster: agentconfig.AgentClusterConfig{CaCertBase64: ""},
73+
Kubelet: agentconfig.AgentKubeletConfig{
74+
ApiServer: srv.URL,
75+
Auth: agentconfig.KubeletAuthInfo{BootstrapToken: "token"},
76+
},
77+
})
78+
79+
results := checker.Check(context.Background())
80+
if len(results) != 1 {
81+
t.Fatalf("len(results) = %d, want 1", len(results))
82+
}
83+
if results[0].Severity != agentpreflight.SeverityOK {
84+
t.Fatalf("severity = %q, want %q: %+v", results[0].Severity, agentpreflight.SeverityOK, results[0])
85+
}
86+
}
87+
88+
func TestAPIServerReachableCheckerInvalidCredentials(t *testing.T) {
89+
t.Parallel()
90+
91+
checker := checkAPIServerReachable(slog.New(slog.DiscardHandler), &agentconfig.AgentConfig{
92+
Cluster: agentconfig.AgentClusterConfig{CaCertBase64: "%%%"},
93+
Kubelet: agentconfig.AgentKubeletConfig{ApiServer: "https://example.test"},
94+
})
95+
96+
results := checker.Check(context.Background())
97+
if len(results) != 1 {
98+
t.Fatalf("len(results) = %d, want 1", len(results))
99+
}
100+
if results[0].Severity != agentpreflight.SeverityError {
101+
t.Fatalf("severity = %q, want %q", results[0].Severity, agentpreflight.SeverityError)
102+
}
103+
if strings.Contains(results[0].Message, "https://example.test") {
104+
t.Fatalf("message leaked API server endpoint: %q", results[0].Message)
105+
}
106+
}

0 commit comments

Comments
 (0)