talm

1
// This Source Code Form is subject to the terms of the Mozilla Public
2
// License, v. 2.0. If a copy of the MPL was not distributed with this
3
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
4

5
// Package machined provides machined implementation.
6
package main
7

8
import (
9
	"context"
10
	"errors"
11
	"fmt"
12
	"log"
13
	"net/http"
14
	"os"
15
	"os/signal"
16
	"path/filepath"
17
	"syscall"
18
	"time"
19

20
	"github.com/hashicorp/go-cleanhttp"
21
	"github.com/siderolabs/go-cmd/pkg/cmd/proc"
22
	"github.com/siderolabs/go-cmd/pkg/cmd/proc/reaper"
23
	debug "github.com/siderolabs/go-debug"
24
	"github.com/siderolabs/go-procfs/procfs"
25
	"golang.org/x/sys/unix"
26

27
	"github.com/aenix-io/talm/internal/app/apid"
28
	"github.com/aenix-io/talm/internal/app/dashboard"
29
	"github.com/aenix-io/talm/internal/app/machined/pkg/runtime"
30
	"github.com/aenix-io/talm/internal/app/machined/pkg/runtime/emergency"
31
	v1alpha1runtime "github.com/aenix-io/talm/internal/app/machined/pkg/runtime/v1alpha1"
32
	"github.com/aenix-io/talm/internal/app/machined/pkg/system"
33
	"github.com/aenix-io/talm/internal/app/machined/pkg/system/services"
34
	"github.com/aenix-io/talm/internal/app/maintenance"
35
	"github.com/aenix-io/talm/internal/app/poweroff"
36
	"github.com/aenix-io/talm/internal/app/trustd"
37
	"github.com/aenix-io/talm/internal/app/wrapperd"
38
	"github.com/aenix-io/talm/internal/pkg/mount"
39
	"github.com/siderolabs/talos/pkg/httpdefaults"
40
	"github.com/siderolabs/talos/pkg/machinery/api/common"
41
	"github.com/siderolabs/talos/pkg/machinery/api/machine"
42
	"github.com/siderolabs/talos/pkg/machinery/constants"
43
	"github.com/siderolabs/talos/pkg/startup"
44
)
45

46
func init() {
47
	// Patch a default HTTP client with updated transport to handle cases when default client is being used.
48
	http.DefaultClient.Transport = httpdefaults.PatchTransport(cleanhttp.DefaultPooledTransport())
49
}
50

51
func recovery(ctx context.Context) {
52
	if r := recover(); r != nil {
53
		var (
54
			err error
55
			ok  bool
56
		)
57

58
		err, ok = r.(error)
59
		if ok {
60
			handle(ctx, err)
61
		}
62
	}
63
}
64

65
// syncNonVolatileStorageBuffers invokes unix.Sync and waits up to 30 seconds
66
// for it to finish.
67
//
68
// See http://man7.org/linux/man-pages/man2/reboot.2.html.
69
func syncNonVolatileStorageBuffers() {
70
	syncdone := make(chan struct{})
71

72
	go func() {
73
		defer close(syncdone)
74

75
		unix.Sync()
76
	}()
77

78
	log.Printf("waiting for sync...")
79

80
	for i := 29; i >= 0; i-- {
81
		select {
82
		case <-syncdone:
83
			log.Printf("sync done")
84

85
			return
86
		case <-time.After(time.Second):
87
		}
88

89
		if i != 0 {
90
			log.Printf("waiting %d more seconds for sync to finish", i)
91
		}
92
	}
93

94
	log.Printf("sync hasn't completed in time, aborting...")
95
}
96

97
//nolint:gocyclo
98
func handle(ctx context.Context, err error) {
99
	rebootCmd := int(emergency.RebootCmd.Load())
100

101
	var rebootErr runtime.RebootError
102

103
	if errors.As(err, &rebootErr) {
104
		// not a failure, but wrapped reboot command
105
		rebootCmd = rebootErr.Cmd
106

107
		err = nil
108
	}
109

110
	if err != nil {
111
		log.Print(err)
112
		revertBootloader(ctx)
113

114
		if p := procfs.ProcCmdline().Get(constants.KernelParamPanic).First(); p != nil {
115
			if *p == "0" {
116
				log.Printf("panic=0 kernel flag found, sleeping forever")
117

118
				rebootCmd = 0
119
			}
120
		}
121
	}
122

123
	if rebootCmd == unix.LINUX_REBOOT_CMD_RESTART {
124
		for i := 10; i >= 0; i-- {
125
			log.Printf("rebooting in %d seconds\n", i)
126
			time.Sleep(1 * time.Second)
127
		}
128
	}
129

130
	if err = proc.KillAll(); err != nil {
131
		log.Printf("error killing all procs: %s", err)
132
	}
133

134
	if err = mount.UnmountAll(); err != nil {
135
		log.Printf("error unmounting: %s", err)
136
	}
137

138
	syncNonVolatileStorageBuffers()
139

140
	if rebootCmd == 0 {
141
		exitSignal := make(chan os.Signal, 1)
142

143
		signal.Notify(exitSignal, syscall.SIGINT, syscall.SIGTERM)
144

145
		<-exitSignal
146
	} else if unix.Reboot(rebootCmd) == nil {
147
		// Wait forever.
148
		select {}
149
	}
150
}
151

152
func runDebugServer(ctx context.Context) {
153
	const debugAddr = ":9982"
154

155
	debugLogFunc := func(msg string) {
156
		log.Print(msg)
157
	}
158

159
	if err := debug.ListenAndServe(ctx, debugAddr, debugLogFunc); err != nil {
160
		log.Fatalf("failed to start debug server: %s", err)
161
	}
162
}
163

164
//nolint:gocyclo
165
func run() error {
166
	errCh := make(chan error)
167

168
	// Limit GOMAXPROCS.
169
	startup.LimitMaxProcs(constants.MachinedMaxProcs)
170

171
	// Set the PATH env var.
172
	if err := os.Setenv("PATH", constants.PATH); err != nil {
173
		return errors.New("error setting PATH")
174
	}
175

176
	// Initialize the controller without a config.
177
	c, err := v1alpha1runtime.NewController()
178
	if err != nil {
179
		return err
180
	}
181

182
	ctx, cancel := context.WithCancel(context.Background())
183
	defer cancel()
184

185
	drainer := runtime.NewDrainer()
186
	defer func() {
187
		drainCtx, drainCtxCancel := context.WithTimeout(context.Background(), time.Second*10)
188
		defer drainCtxCancel()
189

190
		if e := drainer.Drain(drainCtx); e != nil {
191
			log.Printf("WARNING: failed to drain controllers: %s", e)
192
		}
193
	}()
194

195
	go runDebugServer(ctx)
196

197
	// Schedule service shutdown on any return.
198
	defer system.Services(c.Runtime()).Shutdown(ctx)
199

200
	// Start signal and ACPI listeners.
201
	go func() {
202
		if e := c.ListenForEvents(ctx); e != nil {
203
			log.Printf("WARNING: signals and ACPI events will be ignored: %s", e)
204
		}
205
	}()
206

207
	// Start v2 controller runtime.
208
	go func() {
209
		if e := c.V1Alpha2().Run(ctx, drainer); e != nil {
210
			ctrlErr := fmt.Errorf("fatal controller runtime error: %s", e)
211

212
			log.Printf("controller runtime goroutine error: %s", ctrlErr)
213

214
			errCh <- ctrlErr
215
		}
216

217
		log.Printf("controller runtime finished")
218
	}()
219

220
	// Inject controller into maintenance service.
221
	maintenance.InjectController(c)
222

223
	// Load machined service.
224
	system.Services(c.Runtime()).Load(
225
		&services.Machined{Controller: c},
226
	)
227

228
	initializeCanceled := false
229

230
	// Initialize the machine.
231
	if err = c.Run(ctx, runtime.SequenceInitialize, nil); err != nil {
232
		if errors.Is(err, context.Canceled) {
233
			initializeCanceled = true
234
		} else {
235
			return err
236
		}
237
	}
238

239
	// If Initialize sequence was canceled, don't run any other sequence.
240
	if !initializeCanceled {
241
		// Perform an installation if required.
242
		if err = c.Run(ctx, runtime.SequenceInstall, nil); err != nil {
243
			return err
244
		}
245

246
		// Start the machine API.
247
		system.Services(c.Runtime()).LoadAndStart(
248
			&services.APID{},
249
		)
250

251
		// Boot the machine.
252
		if err = c.Run(ctx, runtime.SequenceBoot, nil); err != nil && !errors.Is(err, context.Canceled) {
253
			return err
254
		}
255
	}
256

257
	// Watch and handle runtime events.
258
	//nolint:errcheck
259
	_ = c.Runtime().Events().Watch(
260
		func(events <-chan runtime.EventInfo) {
261
			for {
262
				for event := range events {
263
					switch msg := event.Payload.(type) {
264
					case *machine.SequenceEvent:
265
						if msg.Error != nil {
266
							if msg.Error.GetCode() == common.Code_LOCKED ||
267
								msg.Error.GetCode() == common.Code_CANCELED {
268
								// ignore sequence lock and canceled errors, they're not fatal
269
								continue
270
							}
271

272
							errCh <- fmt.Errorf(
273
								"fatal sequencer error in %q sequence: %v",
274
								msg.GetSequence(),
275
								msg.GetError().String(),
276
							)
277
						}
278
					case *machine.RestartEvent:
279
						errCh <- runtime.RebootError{Cmd: int(msg.Cmd)}
280
					}
281
				}
282
			}
283
		},
284
	)
285

286
	return <-errCh
287
}
288

289
func main() {
290
	ctx, cancel := context.WithCancel(context.Background())
291
	defer cancel()
292

293
	switch filepath.Base(os.Args[0]) {
294
	case "apid":
295
		apid.Main()
296

297
		return
298
	case "trustd":
299
		trustd.Main()
300

301
		return
302
	// Azure uses the hv_utils kernel module to shutdown the node in hyper-v by calling perform_shutdown which will call orderly_poweroff which will call /sbin/poweroff.
303
	case "poweroff", "shutdown":
304
		poweroff.Main(os.Args)
305

306
		return
307
	case "wrapperd":
308
		wrapperd.Main()
309

310
		return
311
	case "dashboard":
312
		dashboard.Main()
313

314
		return
315
	default:
316
	}
317

318
	// Setup panic handler.
319
	defer recovery(ctx)
320

321
	// Initialize the process reaper.
322
	reaper.Run()
323
	defer reaper.Shutdown()
324

325
	handle(ctx, run())
326
}
327