kube play: support auto updates and rollbacks

Add auto-update support to `podman kube play`.  Auto-update policies can
be configured for:
 * the entire pod via the `io.containers.autoupdate` annotation
 * a specific container via the `io.containers.autoupdate/$name` annotation

To make use of rollbacks, the `io.containers.sdnotify` policy should be
set to `container` such that the workload running _inside_ the container
can send the READY message via the NOTIFY_SOCKET once ready.  For
further details on auto updates and rollbacks, please refer to the
specific article [1].

Since auto updates and rollbacks bases on Podman's systemd integration,
the k8s YAML must be executed in the `podman-kube@` systemd template.
For further details on how to run k8s YAML in systemd via Podman, please
refer to the specific article [2].

An examplary k8s YAML may look as follows:
```YAML
apiVersion: v1
kind: Pod
metadata:
  annotations:
      io.containers.autoupdate: "local"
      io.containers.autoupdate/b: "registry"
  labels:
    app: test
  name: test_pod
spec:
  containers:
  - command:
    - top
    image: alpine
    name: a
  - command:
    - top
    image: alpine
    name: b
```

[1] https://www.redhat.com/sysadmin/podman-auto-updates-rollbacks
[2] https://www.redhat.com/sysadmin/kubernetes-workloads-podman-systemd

Signed-off-by: Valentin Rothberg <vrothberg@redhat.com>

2 files changed, 61 insertions, 6 deletions

diff --git a/pkg/systemd/notifyproxy/notifyproxy.go b/pkg/systemd/notifyproxy/notifyproxy.go
index 9e6eb4cf0..1bfab9ca0 100644
--- a/pkg/systemd/notifyproxy/notifyproxy.go
+++ b/pkg/systemd/notifyproxy/notifyproxy.go
@@ -1,12 +1,17 @@
 package notifyproxy
 
 import (
+	"errors"
+	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"os"
 	"strings"
 	"syscall"
+	"time"
 
+	"github.com/containers/podman/v4/libpod/define"
 	"github.com/coreos/go-systemd/v22/daemon"
 	"github.com/sirupsen/logrus"
 )
@@ -39,6 +44,7 @@ func SendMessage(socketPath string, message string) error {
 type NotifyProxy struct {
 	connection *net.UnixConn
 	socketPath string
+	container  Container // optional
 }
 
 // New creates a NotifyProxy.  The specified temp directory can be left empty.
@@ -77,9 +83,26 @@ func (p *NotifyProxy) close() error {
 	return p.connection.Close()
 }
 
+// AddContainer associates a container with the proxy.
+func (p *NotifyProxy) AddContainer(container Container) {
+	p.container = container
+}
+
+// ErrNoReadyMessage is returned when we are waiting for the READY message of a
+// container that is not in the running state anymore.
+var ErrNoReadyMessage = errors.New("container stopped running before READY message was received")
+
+// Container avoids a circular dependency among this package and libpod.
+type Container interface {
+	State() (define.ContainerStatus, error)
+	ID() string
+}
+
 // WaitAndClose waits until receiving the `READY` notify message and close the
 // listener. Note that the this function must only be executed inside a systemd
 // service which will kill the process after a given timeout.
+// If the (optional) container stopped running before the `READY` is received,
+// the waiting gets canceled and ErrNoReadyMessage is returned.
 func (p *NotifyProxy) WaitAndClose() error {
 	defer func() {
 		if err := p.close(); err != nil {
@@ -87,16 +110,48 @@ func (p *NotifyProxy) WaitAndClose() error {
 		}
 	}()
 
+	const bufferSize = 1024
+	sBuilder := strings.Builder{}
 	for {
-		buf := make([]byte, 1024)
-		num, err := p.connection.Read(buf)
-		if err != nil {
+		// Set a read deadline of one second such that we achieve a
+		// non-blocking read and can check if the container has already
+		// stopped running; in that case no READY message will be send
+		// and we're done.
+		if err := p.connection.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
 			return err
 		}
-		for _, s := range strings.Split(string(buf[:num]), "\n") {
-			if s == daemon.SdNotifyReady {
+
+		for {
+			buffer := make([]byte, bufferSize)
+			num, err := p.connection.Read(buffer)
+			if err != nil {
+				if !errors.Is(err, os.ErrDeadlineExceeded) && !errors.Is(err, io.EOF) {
+					return err
+				}
+			}
+			sBuilder.Write(buffer[:num])
+			if num != bufferSize || buffer[num-1] == '\n' {
+				break
+			}
+		}
+
+		for _, line := range strings.Split(sBuilder.String(), "\n") {
+			if line == daemon.SdNotifyReady {
 				return nil
 			}
 		}
+		sBuilder.Reset()
+
+		if p.container == nil {
+			continue
+		}
+
+		state, err := p.container.State()
+		if err != nil {
+			return err
+		}
+		if state != define.ContainerStateRunning {
+			return fmt.Errorf("%w: %s", ErrNoReadyMessage, p.container.ID())
+		}
 	}
 }
diff --git a/pkg/systemd/notifyproxy/notifyproxy_test.go b/pkg/systemd/notifyproxy/notifyproxy_test.go
index ce63fc9cd..066046cb8 100644
--- a/pkg/systemd/notifyproxy/notifyproxy_test.go
+++ b/pkg/systemd/notifyproxy/notifyproxy_test.go
@@ -41,7 +41,7 @@ func TestWaitAndClose(t *testing.T) {
 	default:
 	}
 
-	sendMessage(t, proxy, daemon.SdNotifyReady+"\nsomething else")
+	sendMessage(t, proxy, daemon.SdNotifyReady+"\nsomething else\n")
 	done := func() bool {
 		for i := 0; i < 10; i++ {
 			select {