Add pipe proxy for logs

Author: anahan

cmd/docker-fpm-wrapper/config.go

@@ -20,6 +20,7 @@ type Config struct {
 	FpmNoSlowlogProxy bool `mapstructure:"fpm-no-slowlog"`
 
 	// Logging proxy section
+	WrapperPipe    string `mapstructure:"wrapper-pipe"`
 	WrapperSocket  string `mapstructure:"wrapper-socket"`
 	LineBufferSize int    `mapstructure:"line-buffer-size"`
 
@@ -41,6 +42,7 @@ func parseCommandLineFlags() {
 	pflag.Bool("fpm-no-slowlog", false, "Disable php-fpm slowlog parsing and proxy")
 
 	// Logging proxy section
+	pflag.StringP("wrapper-pipe", "p", "/tmp/fpm-wrapper-pipe", "path to logging pipe, set '' to disable")
 	pflag.StringP("wrapper-socket", "s", "/tmp/fpm-wrapper.sock", "path to logging socket, set null to disable")
 	pflag.Uint("line-buffer-size", 16*1024, "Max log line size (in bytes)")
 

cmd/docker-fpm-wrapper/main.go

@@ -56,14 +56,26 @@ func main() {
 
 	if cfg.WrapperSocket != "null" {
 		env = append(env, fmt.Sprintf("FPM_WRAPPER_SOCK=unix://%s", cfg.WrapperSocket))
-		dataListener := applog.NewSockDataListener(cfg.WrapperSocket, breader.NewPool(cfg.LineBufferSize), syncStderr, errCh)
+		sockDataListener := applog.NewSockDataListener(cfg.WrapperSocket, breader.NewPool(cfg.LineBufferSize), syncStderr, errCh)
 
-		if err = dataListener.Start(); err != nil {
+		if err = sockDataListener.Start(); err != nil {
 			log.Error("Can't start listen", zap.Error(err))
 			os.Exit(1)
 		}
 
-		defer dataListener.Stop()
+		defer sockDataListener.Stop()
+	}
+
+	if cfg.WrapperPipe != "" {
+		env = append(env, fmt.Sprintf("FPM_WRAPPER_PIPE=%s", cfg.WrapperPipe))
+
+		wrapperPipe, err := createFIFOByPathCtx(ctx, cfg.WrapperPipe)
+		if err != nil {
+			log.Error("can't create pipe", zap.Error(err), zap.String("path", cfg.WrapperPipe))
+			os.Exit(1)
+		}
+
+		go applog.NewPipeProxy(log.Named("pipe-proxy"), syncStderr).Proxy(wrapperPipe)
 	}
 
 	fpmConfig, err := phpfpm.ParseConfig(cfg.FpmConfigPath)

internal/applog/data_listener.go

@@ -24,21 +24,6 @@ func NewSockDataListener(sockPath string, rPool *breader.Pool, writer io.Writer,
 	return &SockDataListener{socketPath: sockPath, rPool: rPool, writer: writer, errorChan: errorChan}
 }
 
-func (l *SockDataListener) normalizeLine(line []byte) []byte {
-	ll := len(line)
-	if ll > 0 && line[ll-1] != '\n' {
-		ll += 1
-		line = append(line, '\n')
-	}
-
-	if ll > 1 && line[ll-2] == '\r' {
-		line[ll-2] = line[ll-1]
-		line = line[:ll-1]
-	}
-
-	return line
-}
-
 func (l *SockDataListener) handleConnection(conn net.Conn) {
 	defer conn.Close()
 
@@ -48,7 +33,7 @@ func (l *SockDataListener) handleConnection(conn net.Conn) {
 	for {
 		buf, err := line.ReadOne(reader, true)
 		if len(buf) > 0 {
-			_, _ = l.writer.Write(l.normalizeLine(buf))
+			_, _ = l.writer.Write(normalizeLine(buf))
 		}
 
 		if err == nil {

internal/applog/normalize.go

@@ -0,0 +1,16 @@
+package applog
+
+func normalizeLine(line []byte) []byte {
+	ll := len(line)
+	if ll > 0 && line[ll-1] != '\n' {
+		ll += 1
+		line = append(line, '\n')
+	}
+
+	if ll > 1 && line[ll-2] == '\r' {
+		line[ll-2] = line[ll-1]
+		line = line[:ll-1]
+	}
+
+	return line
+}

internal/applog/pipe_proxy.go

@@ -0,0 +1,41 @@
+package applog
+
+import (
+	"bufio"
+	"errors"
+	"io"
+
+	"go.uber.org/zap"
+
+	"github.com/code-tool/docker-fpm-wrapper/pkg/line"
+)
+
+type PipeProxy struct {
+	log    *zap.Logger
+	writer io.Writer
+}
+
+func NewPipeProxy(log *zap.Logger, writer io.Writer) *PipeProxy {
+	return &PipeProxy{log: log, writer: writer}
+}
+
+func (p *PipeProxy) Proxy(r io.Reader) {
+	bufioReader := bufio.NewReader(r)
+
+	for {
+		buf, err := line.ReadOne(bufioReader, true)
+		if len(buf) > 0 {
+			_, _ = p.writer.Write(normalizeLine(buf))
+		}
+
+		if err == nil {
+			continue
+		}
+
+		if errors.Is(err, io.EOF) {
+			return
+		}
+
+		p.log.Error("can't read line from pipe", zap.Error(err))
+	}
+}