cri-o · openshift-merge-robot · Jul 19, 2021 · Jul 16, 2021 · Jul 9, 2021 · Jul 9, 2021
@@ -47,6 +47,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.10.0
 	github.com/psampaz/go-mod-outdated v0.7.0
+	github.com/rjeczalik/notify v0.9.2
 	github.com/sirupsen/logrus v1.8.1
 	github.com/soheilhy/cmux v0.1.5
 	github.com/stretchr/testify v1.7.0

@@ -1139,6 +1139,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20170806203942-52369c62f446/go.mod h1:uY
 github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/rjeczalik/notify v0.9.2 h1:MiTWrPj55mNDHEiIX5YUSKefw/+lCQVoAFmD6oQm5w8=
+github.com/rjeczalik/notify v0.9.2/go.mod h1:aErll2f0sUX9PXZnVNyeiObbmTlk5jnMoCa4QEjJeqM=
 github.com/robfig/cron v1.1.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
@@ -1519,6 +1521,7 @@ golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180926160741-c2ed4eda69e7/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

@@ -21,10 +21,10 @@ import (
 	"github.com/cri-o/cri-o/server/cri/types"
 	"github.com/cri-o/cri-o/server/metrics"
 	"github.com/cri-o/cri-o/utils"
-	"github.com/fsnotify/fsnotify"
 	json "github.com/json-iterator/go"
 	rspec "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
+	"github.com/rjeczalik/notify"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 	"golang.org/x/sys/unix"
@@ -343,18 +343,41 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
 	cmd.Stdout = &stdoutBuf
 	cmd.Stderr = &stderrBuf
 
-	err = cmd.Start()
+	pidFileCreatedDone := make(chan struct{}, 1)
+	pidFileCreatedCh, err := WatchForFile(pidFile, pidFileCreatedDone, notify.InModify, notify.InMovedTo)
 	if err != nil {
+		return nil, errors.Wrapf(err, "failed to watch %s", pidFile)
+	}
+
+	doneErr := cmd.Start()
+	if doneErr != nil {
 		return nil, err
 	}
 
 	// wait till the command is done
 	done := make(chan error, 1)
 	go func() {
 		done <- cmd.Wait()
+		close(done)
 	}()
 
-	if timeout > 0 {
+	// First, wait for the pid file to be created.
+	// When it is, the timer begins for the exec process.
+	// If the command fails before that happens, however,
+	// that needs to be caught.
+	select {
+	case <-pidFileCreatedCh:
+	case doneErr = <-done:
+	}
+	close(pidFileCreatedDone)
+
+	switch {
+	case doneErr != nil:
+		// If we've already gotten an error from done
+		// the runtime finished before writing the pid file
+		// (probably because the command didn't exist).
+	case timeout > 0:
+		// If there's a timeout, wait for that timeout duration.
 		select {
 		case <-time.After(time.Second * time.Duration(timeout)):
 			// Ensure the process is not left behind
@@ -370,17 +393,18 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
 				Stderr:   []byte(conmonconfig.TimedOutMessage),
 				ExitCode: -1,
 			}, nil
-		case err = <-done:
+		case doneErr = <-done:
 			break
 		}
-	} else {
-		err = <-done
+	default:
+		// If no timeout, just wait until the command finishes.
+		doneErr = <-done
 	}
 
 	// gather exit code from err
 	exitCode := int32(0)
-	if err != nil {
-		if exitError, ok := err.(*exec.ExitError); ok {
+	if doneErr != nil {
+		if exitError, ok := doneErr.(*exec.ExitError); ok {
 			exitCode = int32(exitError.ExitCode())
 		}
 	}
@@ -1017,58 +1041,22 @@ func (r *runtimeOCI) ReopenContainerLog(ctx context.Context, c *Container) error
 	}
 	defer controlFile.Close()
 
-	watcher, err := fsnotify.NewWatcher()
+	done := make(chan struct{}, 1)
+	ch, err := WatchForFile(c.LogPath(), done, notify.InCreate, notify.InModify)
 	if err != nil {
-		return fmt.Errorf("failed to create new watch: %v", err)
-	}
-	defer watcher.Close()
-
-	done := make(chan struct{})
-	doneClosed := false
-	errorCh := make(chan error)
-	go func() {
-		for {
-			select {
-			case event := <-watcher.Events:
-				log.Debugf(ctx, "event: %v", event)
-				if event.Op&fsnotify.Create == fsnotify.Create || event.Op&fsnotify.Write == fsnotify.Write {
-					log.Debugf(ctx, "file created %s", event.Name)
-					if event.Name == c.LogPath() {
-						log.Debugf(ctx, "expected log file created")
-						done <- struct{}{}
-						return
-					}
-				}
-			case err := <-watcher.Errors:
-				errorCh <- fmt.Errorf("watch error for container log reopen %v: %v", c.ID(), err)
-				close(errorCh)
-				return
-			}
-		}
-	}()
-	cLogDir := filepath.Dir(c.LogPath())
-	if err := watcher.Add(cLogDir); err != nil {
-		log.Errorf(ctx, "watcher.Add(%q) failed: %s", cLogDir, err)
-		close(done)
-		doneClosed = true
+		return errors.Wrapf(err, "failed to create watch for %s", c.LogPath())
 	}
 
 	if _, err = fmt.Fprintf(controlFile, "%d %d %d\n", 2, 0, 0); err != nil {
 		log.Debugf(ctx, "Failed to write to control file to reopen log file: %v", err)
 	}
-
 	select {
-	case err := <-errorCh:
-		if !doneClosed {
-			close(done)
-		}
-		return err
-	case <-done:
-		if !doneClosed {
-			close(done)
-		}
-		break
+	case <-ch:
+	case <-time.After(time.Minute * 3):
+		// Give up after 3 minutes, as something wrong probably happened
+		log.Errorf(ctx, "Failed to reopen log file for container %s: timed out", c.ID())
 	}
+	close(done)
 
 	return nil
 }
@@ -1112,3 +1100,34 @@ func prepareProcessExec(c *Container, cmd []string, tty bool) (processFile strin
 func (c *Container) conmonPidFilePath() string {
 	return filepath.Join(c.bundlePath, "conmon-pidfile")
 }
+
+// WatchForFile creates a watch on the parent directory of path, looking for events opsToWatch.
+// It returns immediately with a channel to find when path had one of those events.
+// done can be used to stop the watch.
+// WatchForFile is responsible for closing all internal channels and the returned channel, but not for closing done.
+func WatchForFile(path string, done chan struct{}, opsToWatch ...notify.Event) (chan struct{}, error) {
+	eiCh := make(chan notify.EventInfo, 1)
+	ch := make(chan struct{})
+
+	dir := filepath.Dir(path)
+	if err := notify.Watch(dir, eiCh, opsToWatch...); err != nil {
+		return nil, err
+	}
+	go func() {
+		defer close(ch)
+		defer close(eiCh)
+		defer notify.Stop(eiCh)
+		for {
+			select {
+			case ei := <-eiCh:
+				if ei.Path() == path {
+					ch <- struct{}{}
+					return
+				}
+			case <-done:
+				return
+			}
+		}
+	}()
+	return ch, nil
+}
@@ -3,13 +3,16 @@ package oci_test
 import (
 	"context"
 	"math/rand"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"time"
 
 	"github.com/cri-o/cri-o/internal/oci"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/pkg/errors"
+	"github.com/rjeczalik/notify"
 )
 
 const (
@@ -142,6 +145,71 @@ var _ = t.Describe("Oci", func() {
 			})
 		}
 	})
+	t.Describe("WatchForFile", func() {
+		var notifyFile string
+		var done chan struct{}
+		BeforeEach(func() {
+			notifyFile = filepath.Join(t.MustTempDir("watch"), "file")
+			done = make(chan struct{}, 1)
+		})
+		It("should catch file creation", func() {
+			// Given
+			ch, err := oci.WatchForFile(notifyFile, done, notify.InCreate, notify.InModify)
+			Expect(err).To(BeNil())
+
+			// When
+			f, err := os.Create(notifyFile)
+			Expect(err).To(BeNil())
+			f.Close()
+
+			<-ch
+		})
+		It("should not catch file create if doesn't exist", func() {
+			// Given
+			ch, err := oci.WatchForFile(notifyFile, done, notify.InCreate, notify.InModify)
+			Expect(err).To(BeNil())
+
+			// When
+			f, err := os.Create(notifyFile + "-backup")
+			Expect(err).To(BeNil())
+			f.Close()
+			checkChannelEmpty(ch)
+
+			// Then
+			f, err = os.Create(notifyFile)
+			Expect(err).To(BeNil())
+			f.Close()
+
+			<-ch
+		})
+		It("should only catch file write", func() {
+			// Given
+			ch, err := oci.WatchForFile(notifyFile, done, notify.InModify)
+			Expect(err).To(BeNil())
+
+			// When
+			f, err := os.Create(notifyFile)
+			Expect(err).To(BeNil())
+			defer f.Close()
+
+			checkChannelEmpty(ch)
+
+			_, err = f.Write([]byte("hello"))
+			Expect(err).To(BeNil())
+
+			<-ch
+		})
+		It("should give up after sending on done", func() {
+			// Given
+			ch, err := oci.WatchForFile(notifyFile, done, notify.InModify)
+			Expect(err).To(BeNil())
+
+			// When
+			checkChannelEmpty(ch)
+			done <- struct{}{}
+			<-ch
+		})
+	})
 })
 
 func waitContainerStopAndFailAfterTimeout(ctx context.Context,
@@ -183,3 +251,12 @@ func verifyContainerNotStopped(sut *oci.Container, _ *exec.Cmd, waitError error)
 func inSeconds(d int64) time.Duration {
 	return time.Duration(d) * time.Second
 }
+
+func checkChannelEmpty(ch chan struct{}) {
+	select {
+	case <-ch:
+		// We don't expect to get anything here
+		Expect(true).To(Equal(false))
+	case <-time.After(time.Second * 3):
+	}
+}