add parallel checks

Signed-off-by: Mohammed Naser <mnaser@vexxhost.com>

Author: mnaser

go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	golang.org/x/sync v0.17.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.20.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -35,6 +35,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
 golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=

main.go

@@ -14,7 +14,8 @@ import (
 	"time"
 
 	"github.com/vexxhost/ovsinit/pkg/appctl"
-	"github.com/vexxhost/ovsinit/pkg/succession" // Uses the history.go version
+	"github.com/vexxhost/ovsinit/pkg/succession"
+	"github.com/vexxhost/ovsinit/pkg/verifier"
 )
 
 var (
@@ -186,6 +187,24 @@ func main() {
 			os.Exit(1)
 		}
 
+		verifiers := []verifier.Verifier{
+			verifier.FileRemoval(fmt.Sprintf("%s/%s.pid", appctl.RUN_DIR, binary)),
+			verifier.FileRemoval(fmt.Sprintf("%s/%s.*.ctl", appctl.RUN_DIR, binary)),
+		}
+
+		if binary == "ovs-vswitchd" {
+			verifiers = append(verifiers, verifier.HugePages())
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		err = verifier.Run(ctx, verifiers...)
+		if err != nil {
+			slog.Error("verification after exit failed", "error", err)
+			os.Exit(1)
+		}
+
 		slog.Info("stopped existing process")
 	}
 

pkg/appctl/appctl.go

@@ -5,16 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log/slog"
 	"net"
 	"os"
 	"path/filepath"
-	"time"
 
 	"github.com/cenkalti/rpc2"
 	"github.com/cenkalti/rpc2/jsonrpc"
-	"github.com/fsnotify/fsnotify"
-	"github.com/prometheus/procfs"
 )
 
 const (
@@ -92,104 +88,5 @@ func Cleanup(binary string) error {
 }
 
 func (c *Client) Exit(ctx context.Context, binary string) error {
-	pidPath := fmt.Sprintf("%s/%s.pid", RUN_DIR, binary)
-	ctlPattern := fmt.Sprintf("%s/%s.*.ctl", RUN_DIR, binary)
-
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		return fmt.Errorf("failed to create file watcher: %w", err)
-	}
-	defer watcher.Close()
-
-	err = watcher.Add(RUN_DIR)
-	if err != nil {
-		return fmt.Errorf("failed to watch directory %s: %w", RUN_DIR, err)
-	}
-
-	err = c.CallWithContext(ctx, "exit", []string{}, nil)
-	if err != nil {
-		return fmt.Errorf("failed to send exit command: %w", err)
-	}
-
-	timeout := time.After(30 * time.Second)
-
-	pidRemoved := false
-	ctlRemoved := false
-
-	for {
-		select {
-		case event, ok := <-watcher.Events:
-			if !ok {
-				return errors.New("watcher channel closed")
-			}
-			if event.Op&fsnotify.Remove == fsnotify.Remove {
-				if event.Name == pidPath {
-					pidRemoved = true
-					slog.Info("pid file removed", "pid_file", pidPath)
-				} else if matched, _ := filepath.Match(ctlPattern, event.Name); matched {
-					ctlRemoved = true
-					slog.Info("ctl file removed", "ctl_file", event.Name)
-				}
-
-				if pidRemoved && ctlRemoved {
-					slog.Info("files removed, waiting for hugepages to be available")
-
-					// Get initial hugepages state
-					fs, err := procfs.NewDefaultFS()
-					if err != nil {
-						slog.Warn("cannot access procfs, skipping hugepages check", "error", err)
-						return nil
-					}
-
-					// Poll until hugepages are available
-					ticker := time.NewTicker(50 * time.Millisecond)
-					defer ticker.Stop()
-					hugepagesTimeout := time.After(5 * time.Second)
-
-					for {
-						select {
-						case <-ticker.C:
-							memInfo, err := fs.Meminfo()
-							if err != nil {
-								slog.Warn("cannot read meminfo, assuming process exited", "error", err)
-								return nil
-							}
-
-							// Check if there are free hugepages
-							if memInfo.HugePagesFree != nil && *memInfo.HugePagesFree > 0 {
-								slog.Info("hugepages available, process fully exited",
-									"hugepages_free", *memInfo.HugePagesFree)
-								return nil
-							}
-
-							// Also check if no hugepages are configured
-							if memInfo.HugePagesTotal != nil && *memInfo.HugePagesTotal == 0 {
-								slog.Info("no hugepages configured, process fully exited")
-								return nil
-							}
-
-							slog.Debug("waiting for hugepages to be freed",
-								"hugepages_total", memInfo.HugePagesTotal,
-								"hugepages_free", memInfo.HugePagesFree)
-
-						case <-hugepagesTimeout:
-							slog.Warn("timeout waiting for hugepages, proceeding anyway")
-							return nil
-						case <-ctx.Done():
-							return ctx.Err()
-						}
-					}
-				}
-			}
-		case err, ok := <-watcher.Errors:
-			if !ok {
-				return errors.New("watcher error channel closed")
-			}
-			return fmt.Errorf("watcher error: %w", err)
-		case <-timeout:
-			return fmt.Errorf("timeout waiting for process to exit (pid_removed=%v, ctl_removed=%v)", pidRemoved, ctlRemoved)
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
+	return c.CallWithContext(ctx, "exit", []string{}, nil)
 }

pkg/verifier/file_removal.go

@@ -0,0 +1,80 @@
+package verifier
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"path/filepath"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+type FileRemovalVerifier struct {
+	pattern string
+}
+
+func FileRemoval(pattern string) *FileRemovalVerifier {
+	return &FileRemovalVerifier{
+		pattern: pattern,
+	}
+}
+
+func (v *FileRemovalVerifier) String() string {
+	return fmt.Sprintf("file_removal(%s)", v.pattern)
+}
+
+func (v *FileRemovalVerifier) Verify(ctx context.Context) error {
+	// Check if files matching the pattern exist
+	matches, err := filepath.Glob(v.pattern)
+	if err != nil {
+		return fmt.Errorf("failed to check pattern %s: %w", v.pattern, err)
+	}
+
+	if len(matches) == 0 {
+		slog.Info(fmt.Sprintf("%s: no files found, already removed", v.String()))
+		return nil
+	}
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return fmt.Errorf("failed to create watcher: %w", err)
+	}
+	defer func() {
+		if err := watcher.Close(); err != nil {
+			slog.Warn("failed to close watcher", "error", err)
+		}
+	}()
+
+	dir := filepath.Dir(v.pattern)
+	if err := watcher.Add(dir); err != nil {
+		return fmt.Errorf("failed to watch directory %s: %w", dir, err)
+	}
+
+	for {
+		select {
+		case event, ok := <-watcher.Events:
+			if !ok {
+				return errors.New("watcher channel closed")
+			}
+
+			if event.Op&fsnotify.Remove == fsnotify.Remove {
+				matches, _ := filepath.Match(v.pattern, event.Name)
+
+				if matches {
+					slog.Info(fmt.Sprintf("%s: file removed", v.String()),
+						"file", event.Name)
+					return nil
+				}
+			}
+		case err, ok := <-watcher.Errors:
+			if !ok {
+				return errors.New("watcher error channel closed")
+			}
+
+			return fmt.Errorf("watcher error: %w", err)
+		case <-ctx.Done():
+			return fmt.Errorf("timeout waiting for %s: %w", v.String(), ctx.Err())
+		}
+	}
+}

pkg/verifier/file_removal_test.go

@@ -0,0 +1,107 @@
+package verifier
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFileRemovalVerifier(t *testing.T) {
+	tempDir := t.TempDir()
+
+	// Test exact path matching
+	testFile := filepath.Join(tempDir, "test.txt")
+	err := os.WriteFile(testFile, []byte("test"), 0644)
+	require.NoError(t, err)
+
+	verifier := FileRemoval(testFile)
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		err = os.Remove(testFile)
+		require.NoError(t, err)
+	}()
+
+	err = verifier.Verify(t.Context())
+	assert.NoError(t, err)
+}
+
+func TestFileRemovalVerifierWithGlob(t *testing.T) {
+	tempDir := t.TempDir()
+
+	testFile := filepath.Join(tempDir, "test.12345.log")
+	err := os.WriteFile(testFile, []byte("test"), 0644)
+	require.NoError(t, err)
+
+	pattern := filepath.Join(tempDir, "test.*.log")
+	verifier := FileRemoval(pattern)
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		err := os.Remove(testFile)
+		require.NoError(t, err)
+	}()
+
+	err = verifier.Verify(t.Context())
+	assert.NoError(t, err)
+}
+
+func TestFileRemovalVerifier_AlreadyRemoved(t *testing.T) {
+	tempDir := t.TempDir()
+
+	pattern := filepath.Join(tempDir, "nonexistent.txt")
+	verifier := FileRemoval(pattern)
+
+	err := verifier.Verify(t.Context())
+	assert.NoError(t, err)
+}
+
+func TestFileRemovalVerifier_AlreadyRemovedGlob(t *testing.T) {
+	tempDir := t.TempDir()
+
+	pattern := filepath.Join(tempDir, "*.nonexistent")
+	verifier := FileRemoval(pattern)
+
+	err := verifier.Verify(t.Context())
+	assert.NoError(t, err)
+}
+
+func TestFileRemovalVerifier_Timeout(t *testing.T) {
+	tempDir := t.TempDir()
+
+	testFile := filepath.Join(tempDir, "persistent.txt")
+	err := os.WriteFile(testFile, []byte("test"), 0644)
+	require.NoError(t, err)
+
+	verifier := FileRemoval(testFile)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Millisecond)
+	defer cancel()
+
+	err = verifier.Verify(ctx)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "timeout")
+}
+
+func TestFileRemovalVerifier_ContextCancellation(t *testing.T) {
+	tempDir := t.TempDir()
+
+	testFile := filepath.Join(tempDir, "persistent.txt")
+	err := os.WriteFile(testFile, []byte("test"), 0644)
+	require.NoError(t, err)
+
+	verifier := FileRemoval(testFile)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		cancel()
+	}()
+
+	err = verifier.Verify(ctx)
+	assert.Error(t, err)
+	assert.ErrorIs(t, err, context.Canceled)
+}