Skip to content

Commit 9a7d193

Browse files
hiroTamadahiroTamada
authored
Merge branch 'main' into hypeship/fix-tc-class-leak
2 parents 4ea5f81 + c2e2295 commit 9a7d193

12 files changed

Lines changed: 285 additions & 64 deletions

File tree

‎lib/instances/fork.go‎

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
restartpolicy "github.com/kernel/hypeman/lib/restart-policy"
2020
"github.com/nrednav/cuid2"
2121
"go.opentelemetry.io/otel/attribute"
22+
"go.opentelemetry.io/otel/trace"
2223
"gvisor.dev/gvisor/pkg/cleanup"
2324
)
2425

@@ -264,7 +265,7 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin
264265
return nil, false, err
265266
}
266267

267-
existsByMetadata, err := m.instanceNameExists(req.Name)
268+
existsByMetadata, err := m.instanceNameExists(ctx, req.Name, "fork_precheck")
268269
if err != nil {
269270
return nil, false, fmt.Errorf("check instance name availability: %w", err)
270271
}
@@ -422,31 +423,51 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin
422423
}
423424

424425
func (m *manager) saveForkMetadata(ctx context.Context, meta *metadata) error {
426+
ctx, span := m.tracerOrDefault().Start(ctx, "instances.fork_metadata.save",
427+
trace.WithAttributes(attribute.String("operation", "save_fork_metadata")),
428+
)
429+
var retErr error
430+
defer func() { finishInstancesSpan(span, retErr) }()
431+
432+
lockCtx, lockWaitSpan := m.tracerOrDefault().Start(ctx, "instances.fork_metadata.lock_wait",
433+
trace.WithAttributes(attribute.String("operation", "fork_metadata_lock_wait")),
434+
)
425435
m.forkMetadataMu.Lock()
436+
finishInstancesSpan(lockWaitSpan, nil)
437+
438+
holdCtx, lockHoldSpan := m.tracerOrDefault().Start(lockCtx, "instances.fork_metadata.lock_hold",
439+
trace.WithAttributes(attribute.String("operation", "fork_metadata_lock_hold")),
440+
)
426441
defer m.forkMetadataMu.Unlock()
442+
defer func() { finishInstancesSpan(lockHoldSpan, retErr) }()
427443

428444
// Earlier name checks are advisory so callers can fail before doing fork
429445
// work when possible. This is the serialized admission point for fork
430446
// metadata, so concurrent forks re-check names immediately before save.
431447
name := meta.Name
432-
existsByMetadata, err := m.instanceNameExists(name)
448+
existsByMetadata, err := m.instanceNameExists(holdCtx, name, "fork_admission")
433449
if err != nil {
434-
return fmt.Errorf("check instance name availability: %w", err)
450+
retErr = fmt.Errorf("check instance name availability: %w", err)
451+
return retErr
435452
}
436453
if existsByMetadata {
437-
return fmt.Errorf("%w: instance name '%s' already exists", ErrAlreadyExists, name)
454+
retErr = fmt.Errorf("%w: instance name '%s' already exists", ErrAlreadyExists, name)
455+
return retErr
438456
}
439457
if meta.NetworkEnabled {
440-
exists, err := m.networkManager.NameExists(ctx, name, "")
458+
exists, err := m.networkManager.NameExists(holdCtx, name, "")
441459
if err != nil {
442-
return fmt.Errorf("check instance name availability: %w", err)
460+
retErr = fmt.Errorf("check instance name availability: %w", err)
461+
return retErr
443462
}
444463
if exists {
445-
return fmt.Errorf("%w: instance name '%s' already exists in network", ErrAlreadyExists, name)
464+
retErr = fmt.Errorf("%w: instance name '%s' already exists in network", ErrAlreadyExists, name)
465+
return retErr
446466
}
447467
}
448468
if err := m.saveMetadata(meta); err != nil {
449-
return fmt.Errorf("save fork metadata: %w", err)
469+
retErr = fmt.Errorf("save fork metadata: %w", err)
470+
return retErr
450471
}
451472
return nil
452473
}

‎lib/instances/query.go‎

Lines changed: 32 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"github.com/kernel/hypeman/lib/instances/phasetracking"
2121
"github.com/kernel/hypeman/lib/logger"
2222
"go.opentelemetry.io/otel/attribute"
23+
"go.opentelemetry.io/otel/trace"
2324
)
2425

2526
// exitSentinelPrefix is the machine-parseable prefix written by init to serial console.
@@ -879,22 +880,39 @@ func (m *manager) listInstances(ctx context.Context) ([]Instance, error) {
879880
return result, nil
880881
}
881882

882-
func (m *manager) findInstanceMetadataByExactName(name string) (*metadata, error) {
883+
func (m *manager) findInstanceMetadataByExactName(ctx context.Context, name string) (*metadata, error) {
884+
ctx, span := m.tracerOrDefault().Start(ctx, "instances.metadata.find_exact_name",
885+
trace.WithAttributes(attribute.String("operation", "find_exact_name")),
886+
)
887+
defer span.End()
888+
883889
files, err := m.listMetadataFiles()
884890
if err != nil {
891+
span.RecordError(err)
885892
return nil, err
886893
}
894+
span.SetAttributes(attribute.Int("metadata_files", len(files)))
887895

896+
scanned := 0
888897
for _, file := range files {
889898
id := filepath.Base(filepath.Dir(file))
899+
scanned++
890900
meta, err := m.loadMetadata(id)
891901
if err != nil {
892902
continue
893903
}
894904
if meta.Name == name {
905+
span.SetAttributes(
906+
attribute.Int("metadata_files_scanned", scanned),
907+
attribute.Bool("matched", true),
908+
)
895909
return meta, nil
896910
}
897911
}
912+
span.SetAttributes(
913+
attribute.Int("metadata_files_scanned", scanned),
914+
attribute.Bool("matched", false),
915+
)
898916
return nil, ErrNotFound
899917
}
900918

@@ -949,14 +967,25 @@ func (m *manager) findInstanceMetadataByNameOrIDPrefix(idOrName string, minPrefi
949967
return nil, ErrNotFound
950968
}
951969

952-
func (m *manager) instanceNameExists(name string) (bool, error) {
953-
_, err := m.findInstanceMetadataByExactName(name)
970+
func (m *manager) instanceNameExists(ctx context.Context, name, caller string) (bool, error) {
971+
ctx, span := m.tracerOrDefault().Start(ctx, "instances.metadata.name_exists",
972+
trace.WithAttributes(
973+
attribute.String("operation", "metadata_name_exists"),
974+
attribute.String("caller", caller),
975+
),
976+
)
977+
defer span.End()
978+
979+
_, err := m.findInstanceMetadataByExactName(ctx, name)
954980
if err == nil {
981+
span.SetAttributes(attribute.Bool("exists", true))
955982
return true, nil
956983
}
957984
if err == ErrNotFound {
985+
span.SetAttributes(attribute.Bool("exists", false))
958986
return false, nil
959987
}
988+
span.RecordError(err)
960989
return false, err
961990
}
962991

‎lib/instances/snapshot.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -625,7 +625,7 @@ func (m *manager) ensureSnapshotNameAvailable(sourceInstanceID, snapshotName str
625625
}
626626

627627
func (m *manager) ensureInstanceNameAvailableForSnapshotFork(ctx context.Context, name string, networkEnabled bool) error {
628-
existsByMetadata, err := m.instanceNameExists(name)
628+
existsByMetadata, err := m.instanceNameExists(ctx, name, "snapshot_create")
629629
if err != nil {
630630
return fmt.Errorf("check instance name availability: %w", err)
631631
}

‎lib/instances/snapshot_alias_lock.go‎

Lines changed: 73 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@ import (
77

88
"github.com/kernel/hypeman/lib/forkvm"
99
"github.com/kernel/hypeman/lib/hypervisor"
10+
"go.opentelemetry.io/otel"
11+
"go.opentelemetry.io/otel/attribute"
12+
"go.opentelemetry.io/otel/trace"
1013
)
1114

1215
func withSnapshotSourceAliasReadLock(run func() error) error {
@@ -16,13 +19,19 @@ func withSnapshotSourceAliasReadLock(run func() error) error {
1619
}
1720

1821
func prepareForkWithAliasReadLock(ctx context.Context, starter hypervisor.VMStarter, req hypervisor.ForkPrepareRequest) (hypervisor.ForkPrepareResult, error) {
22+
ctx, span := startInstancesSpan(ctx, otel.Tracer("hypeman/instances"), "instances.snapshot_alias.prepare_fork",
23+
attribute.String("operation", "snapshot_alias_prepare_fork"),
24+
)
25+
var retErr error
26+
defer func() { finishInstancesSpan(span, retErr) }()
27+
1928
var result hypervisor.ForkPrepareResult
20-
err := withSnapshotSourceAliasReadLock(func() error {
29+
retErr = withSnapshotSourceAliasReadLock(func() error {
2130
var err error
2231
result, err = starter.PrepareFork(ctx, req)
2332
return err
2433
})
25-
return result, err
34+
return result, retErr
2635
}
2736

2837
func copyGuestDirectoryWithAliasReadLock(srcDir, dstDir string) error {
@@ -32,16 +41,42 @@ func copyGuestDirectoryWithAliasReadLock(srcDir, dstDir string) error {
3241
}
3342

3443
func (m *manager) copyForkSourceGuestDirectory(ctx context.Context, sourceState State, sourceID string, stored *StoredMetadata, srcDir, dstDir, deferredSnapshotMemoryPath string) error {
44+
ctx, span := m.tracerOrDefault().Start(ctx, "instances.fork.copy_guest_directory",
45+
trace.WithAttributes(
46+
attribute.String("operation", "fork_copy_guest_directory"),
47+
attribute.String("instance_id", sourceID),
48+
attribute.String("source_state", string(sourceState)),
49+
attribute.Bool("deferred_snapshot_memory", deferredSnapshotMemoryPath != ""),
50+
),
51+
)
52+
var retErr error
53+
defer func() { finishInstancesSpan(span, retErr) }()
54+
3555
if sourceState == StateStandby {
36-
if err := m.ensureSnapshotMemoryReady(ctx, m.paths.InstanceSnapshotLatest(sourceID), m.snapshotJobKeyForInstance(sourceID), stored.HypervisorType); err != nil {
37-
return fmt.Errorf("prepare standby snapshot for fork: %w", err)
56+
readyCtx, readyDone := m.startLifecycleStep(ctx, "instances.fork.copy_guest_directory.ensure_snapshot_memory_ready",
57+
attribute.String("operation", "fork_copy_ensure_snapshot_memory_ready"),
58+
attribute.String("instance_id", sourceID),
59+
attribute.String("hypervisor", string(stored.HypervisorType)),
60+
)
61+
if err := m.ensureSnapshotMemoryReady(readyCtx, m.paths.InstanceSnapshotLatest(sourceID), m.snapshotJobKeyForInstance(sourceID), stored.HypervisorType); err != nil {
62+
readyDone(err)
63+
retErr = fmt.Errorf("prepare standby snapshot for fork: %w", err)
64+
return retErr
3865
}
66+
readyDone(nil)
3967
}
68+
4069
copyOptions := forkvm.CopyOptions{}
4170
if deferredSnapshotMemoryPath != "" {
4271
copyOptions.SkipRelativePaths = map[string]struct{}{firecrackerSnapshotMemoryRelPath: {}}
4372
}
44-
return withSnapshotSourceAliasReadLock(func() error {
73+
74+
_, cloneDone := m.startLifecycleStep(ctx, "instances.fork.copy_guest_directory.clone",
75+
attribute.String("operation", "fork_copy_guest_directory_clone"),
76+
attribute.String("instance_id", sourceID),
77+
attribute.Bool("deferred_snapshot_memory", deferredSnapshotMemoryPath != ""),
78+
)
79+
retErr = withSnapshotSourceAliasReadLock(func() error {
4580
if err := forkvm.CopyGuestDirectoryWithOptions(srcDir, dstDir, copyOptions); err != nil {
4681
if errors.Is(err, forkvm.ErrSparseCopyUnsupported) {
4782
return fmt.Errorf("fork requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err)
@@ -50,17 +85,45 @@ func (m *manager) copyForkSourceGuestDirectory(ctx context.Context, sourceState
5085
}
5186
return nil
5287
})
88+
cloneDone(retErr)
89+
return retErr
5390
}
5491

5592
func (m *manager) copySnapshotGuestDirectoryForFork(ctx context.Context, snapshotID string, hvType hypervisor.Type, dstDir, deferredSnapshotMemoryPath string) error {
56-
if err := m.ensureSnapshotMemoryReady(ctx, m.paths.SnapshotGuestDir(snapshotID), "", hvType); err != nil {
57-
return fmt.Errorf("prepare snapshot memory for fork: %w", err)
93+
ctx, span := m.tracerOrDefault().Start(ctx, "instances.snapshot.copy_guest_directory",
94+
trace.WithAttributes(
95+
attribute.String("operation", "snapshot_copy_guest_directory"),
96+
attribute.String("snapshot_id", snapshotID),
97+
attribute.String("hypervisor", string(hvType)),
98+
attribute.Bool("deferred_snapshot_memory", deferredSnapshotMemoryPath != ""),
99+
),
100+
)
101+
var retErr error
102+
defer func() { finishInstancesSpan(span, retErr) }()
103+
104+
readyCtx, readyDone := m.startLifecycleStep(ctx, "instances.snapshot.copy_guest_directory.ensure_snapshot_memory_ready",
105+
attribute.String("operation", "snapshot_copy_ensure_snapshot_memory_ready"),
106+
attribute.String("snapshot_id", snapshotID),
107+
attribute.String("hypervisor", string(hvType)),
108+
)
109+
if err := m.ensureSnapshotMemoryReady(readyCtx, m.paths.SnapshotGuestDir(snapshotID), "", hvType); err != nil {
110+
readyDone(err)
111+
retErr = fmt.Errorf("prepare snapshot memory for fork: %w", err)
112+
return retErr
58113
}
114+
readyDone(nil)
115+
59116
copyOptions := forkvm.CopyOptions{}
60117
if deferredSnapshotMemoryPath != "" {
61118
copyOptions.SkipRelativePaths = map[string]struct{}{firecrackerSnapshotMemoryRelPath: {}}
62119
}
63-
return withSnapshotSourceAliasReadLock(func() error {
120+
121+
_, cloneDone := m.startLifecycleStep(ctx, "instances.snapshot.copy_guest_directory.clone",
122+
attribute.String("operation", "snapshot_copy_guest_directory_clone"),
123+
attribute.String("snapshot_id", snapshotID),
124+
attribute.Bool("deferred_snapshot_memory", deferredSnapshotMemoryPath != ""),
125+
)
126+
retErr = withSnapshotSourceAliasReadLock(func() error {
64127
if err := forkvm.CopyGuestDirectoryWithOptions(m.paths.SnapshotGuestDir(snapshotID), dstDir, copyOptions); err != nil {
65128
if errors.Is(err, forkvm.ErrSparseCopyUnsupported) {
66129
return fmt.Errorf("fork from snapshot requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err)
@@ -69,4 +132,6 @@ func (m *manager) copySnapshotGuestDirectoryForFork(ctx context.Context, snapsho
69132
}
70133
return nil
71134
})
135+
cloneDone(retErr)
136+
return retErr
72137
}

‎lib/instances/standby.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -369,6 +369,11 @@ func (m *manager) shutdownHypervisor(ctx context.Context, inst *Instance) error
369369
shutdownErr = hv.Shutdown(ctx)
370370
}
371371

372+
// Teardown is committed; prevent new control-socket clients while the
373+
// hypervisor exits. The deferred remove remains as a fallback for early
374+
// returns above.
375+
_ = os.Remove(inst.SocketPath)
376+
372377
// Wait for process to exit
373378
if inst.HypervisorPID != nil {
374379
pid := *inst.HypervisorPID

‎lib/instances/standby_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,47 @@
11
package instances
22

33
import (
4+
"net"
45
"os"
56
"path/filepath"
67
"testing"
78

9+
"github.com/kernel/hypeman/lib/hypervisor"
810
"github.com/stretchr/testify/assert"
911
"github.com/stretchr/testify/require"
1012
)
1113

14+
func TestShutdownHypervisorRemovesControlSocket(t *testing.T) {
15+
tmpDir, err := os.MkdirTemp("/tmp", "hypeman-standby-socket-")
16+
require.NoError(t, err)
17+
t.Cleanup(func() {
18+
_ = os.RemoveAll(tmpDir)
19+
})
20+
socketPath := filepath.Join(tmpDir, "noop.sock")
21+
listener, err := net.Listen("unix", socketPath)
22+
require.NoError(t, err)
23+
require.NoError(t, listener.Close())
24+
25+
lifecycleNoopHypervisorStates.Store(socketPath, hypervisor.StateRunning)
26+
t.Cleanup(func() {
27+
lifecycleNoopHypervisorStates.Delete(socketPath)
28+
})
29+
30+
m := &manager{}
31+
inst := &Instance{
32+
StoredMetadata: StoredMetadata{
33+
Id: "standby-socket-cleanup",
34+
SocketPath: socketPath,
35+
HypervisorType: lifecycleNoopHypervisorType,
36+
},
37+
}
38+
39+
require.NoError(t, m.shutdownHypervisor(t.Context(), inst))
40+
41+
_, err = os.Stat(socketPath)
42+
require.True(t, os.IsNotExist(err), "shutdown should remove the hypervisor control socket")
43+
}
44+
1245
func TestDiscardPromotedRetainedSnapshotTargetAfterSnapshotError(t *testing.T) {
1346
t.Parallel()
1447

‎lib/uffdpager/VERSION‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
0.1.2
1+
0.1.3

0 commit comments

Comments
 (0)