1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
|
package notifyproxy
import (
"context"
"errors"
"fmt"
"io"
"net"
"os"
"strings"
"syscall"
"time"
"github.com/containers/podman/v4/libpod/define"
"github.com/coreos/go-systemd/v22/daemon"
"github.com/sirupsen/logrus"
)
// SendMessage sends the specified message to the specified socket.
// No message is sent if no socketPath is provided and the NOTIFY_SOCKET
// variable is not set either.
func SendMessage(socketPath string, message string) error {
if socketPath == "" {
socketPath, _ = os.LookupEnv("NOTIFY_SOCKET")
if socketPath == "" {
return nil
}
}
socketAddr := &net.UnixAddr{
Name: socketPath,
Net: "unixgram",
}
conn, err := net.DialUnix(socketAddr.Net, nil, socketAddr)
if err != nil {
return err
}
defer conn.Close()
_, err = conn.Write([]byte(message))
return err
}
// NotifyProxy can be used to proxy notify messages.
type NotifyProxy struct {
connection *net.UnixConn
socketPath string
container Container // optional
}
// New creates a NotifyProxy. The specified temp directory can be left empty.
func New(tmpDir string) (*NotifyProxy, error) {
tempFile, err := os.CreateTemp(tmpDir, "-podman-notify-proxy.sock")
if err != nil {
return nil, err
}
defer tempFile.Close()
socketPath := tempFile.Name()
if err := syscall.Unlink(socketPath); err != nil { // Unlink the socket so we can bind it
return nil, err
}
socketAddr := &net.UnixAddr{
Name: socketPath,
Net: "unixgram",
}
conn, err := net.ListenUnixgram(socketAddr.Net, socketAddr)
if err != nil {
return nil, err
}
return &NotifyProxy{connection: conn, socketPath: socketPath}, nil
}
// SocketPath returns the path of the socket the proxy is listening on.
func (p *NotifyProxy) SocketPath() string {
return p.socketPath
}
// close closes the listener and removes the socket.
func (p *NotifyProxy) close() error {
defer os.Remove(p.socketPath)
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 {
logrus.Errorf("Closing notify proxy: %v", err)
}
}()
// Since reading from the connection is blocking, we need to spin up two
// goroutines. One waiting for the `READY` message, the other waiting
// for the container to stop running.
errorChan := make(chan error, 1)
readyChan := make(chan bool, 1)
go func() {
// Read until the `READY` message is received or the connection
// is closed.
const bufferSize = 1024
sBuilder := strings.Builder{}
for {
for {
buffer := make([]byte, bufferSize)
num, err := p.connection.Read(buffer)
if err != nil {
if !errors.Is(err, io.EOF) {
errorChan <- err
return
}
}
sBuilder.Write(buffer[:num])
if num != bufferSize || buffer[num-1] == '\n' {
// Break as we read an entire line that
// we can inspect for the `READY`
// message.
break
}
}
for _, line := range strings.Split(sBuilder.String(), "\n") {
if line == daemon.SdNotifyReady {
readyChan <- true
return
}
}
sBuilder.Reset()
}
}()
if p.container != nil {
// Create a cancellable context to make sure the goroutine
// below terminates.
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
select {
case <-ctx.Done():
return
default:
state, err := p.container.State()
if err != nil {
errorChan <- err
return
}
if state != define.ContainerStateRunning {
errorChan <- fmt.Errorf("%w: %s", ErrNoReadyMessage, p.container.ID())
return
}
time.Sleep(time.Second)
}
}()
}
// Wait for the ready/error channel.
select {
case <-readyChan:
return nil
case err := <-errorChan:
return err
}
}
|
