Monitor NGINX logs for errors & NGINX worker processes after a NGINX reload

src/core/config/config.go

@@ -78,6 +78,7 @@ func SetDefaults() {
 
 	// NGINX DEFAULTS
 	Viper.SetDefault(NginxClientVersion, Defaults.Nginx.NginxClientVersion)
+	Viper.SetDefault(NginxConfigReloadMonitoringPeriod, Defaults.Nginx.ConfigReloadMonitoringPeriod)
 }
 
 func setFlagDeprecated(name string, usageMessage string) {
@@ -284,10 +285,11 @@ func getDataplane() Dataplane {
 
 func getNginx() Nginx {
 	return Nginx{
-		ExcludeLogs:         Viper.GetString(NginxExcludeLogs),
-		Debug:               Viper.GetBool(NginxDebug),
-		NginxCountingSocket: Viper.GetString(NginxCountingSocket),
-		NginxClientVersion:  Viper.GetInt(NginxClientVersion),
+		ExcludeLogs:                  Viper.GetString(NginxExcludeLogs),
+		Debug:                        Viper.GetBool(NginxDebug),
+		NginxCountingSocket:          Viper.GetString(NginxCountingSocket),
+		NginxClientVersion:           Viper.GetInt(NginxClientVersion),
+		ConfigReloadMonitoringPeriod: Viper.GetDuration(NginxConfigReloadMonitoringPeriod),
 	}
 }
 

src/core/config/config_test.go

@@ -148,6 +148,7 @@ func TestGetConfig(t *testing.T) {
 		assert.Equal(t, Defaults.Server.Metrics, config.Server.Metrics)
 
 		assert.Equal(t, Defaults.AgentAPI.Port, config.AgentAPI.Port)
+		assert.Equal(t, Defaults.Nginx.ConfigReloadMonitoringPeriod, config.Nginx.ConfigReloadMonitoringPeriod)
 
 		assert.True(t, len(config.AllowedDirectoriesMap) > 0)
 		assert.Equal(t, Defaults.ConfigDirs, config.ConfigDirs)

src/core/config/defaults.go

@@ -51,9 +51,10 @@ var (
 			Token: uuid.New().String(),
 		},
 		Nginx: Nginx{
-			Debug:               false,
-			NginxCountingSocket: "unix:/var/run/nginx-agent/nginx.sock",
-			NginxClientVersion:  6,
+			Debug:                        false,
+			NginxCountingSocket:          "unix:/var/run/nginx-agent/nginx.sock",
+			NginxClientVersion:           6,
+			ConfigReloadMonitoringPeriod: 10 * time.Second,
 		},
 		ConfigDirs:            "/etc/nginx:/usr/local/etc/nginx:/usr/share/nginx/modules:/etc/nms",
 		AllowedDirectoriesMap: map[string]struct{}{},
@@ -128,10 +129,11 @@ const (
 	// viper keys used in config
 	NginxKey = "nginx"
 
-	NginxExcludeLogs    = NginxKey + agent_config.KeyDelimiter + "exclude_logs"
-	NginxDebug          = NginxKey + agent_config.KeyDelimiter + "debug"
-	NginxCountingSocket = NginxKey + agent_config.KeyDelimiter + "socket"
-	NginxClientVersion  = NginxKey + agent_config.KeyDelimiter + "client_version"
+	NginxExcludeLogs                  = NginxKey + agent_config.KeyDelimiter + "exclude_logs"
+	NginxDebug                        = NginxKey + agent_config.KeyDelimiter + "debug"
+	NginxCountingSocket               = NginxKey + agent_config.KeyDelimiter + "socket"
+	NginxClientVersion                = NginxKey + agent_config.KeyDelimiter + "client_version"
+	NginxConfigReloadMonitoringPeriod = NginxKey + agent_config.KeyDelimiter + "config_reload_monitoring_period"
 
 	// viper keys used in config
 	DataplaneKey = "dataplane"
@@ -247,6 +249,11 @@ var (
 			Usage:        "The NGINX Plus counting unix socket location.",
 			DefaultValue: Defaults.Nginx.NginxCountingSocket,
 		},
+		&DurationFlag{
+			Name:         NginxConfigReloadMonitoringPeriod,
+			Usage:        "The period at which the NGINX Agent will monitor error logs after a NGINX reload.",
+			DefaultValue: Defaults.Nginx.ConfigReloadMonitoringPeriod,
+		},
 		// Metrics
 		&DurationFlag{
 			Name:         MetricsCollectionInterval,

src/core/config/types.go

@@ -79,10 +79,11 @@ type TLSConfig struct {
 
 // Nginx settings
 type Nginx struct {
-	ExcludeLogs         string `mapstructure:"exclude_logs" yaml:"-"`
-	Debug               bool   `mapstructure:"debug" yaml:"-"`
-	NginxCountingSocket string `mapstructure:"socket" yaml:"-"`
-	NginxClientVersion  int    `mapstructure:"client_version" yaml:"-"`
+	ExcludeLogs                  string        `mapstructure:"exclude_logs" yaml:"-"`
+	Debug                        bool          `mapstructure:"debug" yaml:"-"`
+	NginxCountingSocket          string        `mapstructure:"socket" yaml:"-"`
+	NginxClientVersion           int           `mapstructure:"client_version" yaml:"-"`
+	ConfigReloadMonitoringPeriod time.Duration `mapstructure:"config_reload_monitoring_period" yaml:"-"`
 }
 
 type Dataplane struct {

src/core/metrics/sources/nginx_access_log.go

@@ -20,7 +20,7 @@ import (
 	"github.com/nginx/agent/sdk/v2/proto"
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/metrics"
-	"github.com/nginx/agent/v2/src/core/metrics/sources/tailer"
+	"github.com/nginx/agent/v2/src/core/tailer"
 
 	log "github.com/sirupsen/logrus"
 )

src/core/metrics/sources/nginx_error_log.go

@@ -16,7 +16,7 @@ import (
 	"github.com/nginx/agent/sdk/v2/proto"
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/metrics"
-	"github.com/nginx/agent/v2/src/core/metrics/sources/tailer"
+	"github.com/nginx/agent/v2/src/core/tailer"
 
 	log "github.com/sirupsen/logrus"
 )

src/core/metrics/sources/tailer/tailer.go → src/core/tailer/tailer.go

@@ -19,9 +19,10 @@ import (
 
 var (
 	tailConfig = tail.Config{
-		Follow: true,
-		ReOpen: true,
-		Poll:   true,
+		Follow:    true,
+		ReOpen:    true,
+		MustExist: true,
+		Poll:      true,
 		Location: &tail.SeekInfo{
 			Whence: io.SeekEnd,
 		},
@@ -107,11 +108,14 @@ func (t *Tailer) Tail(ctx context.Context, data chan<- string) {
 			data <- line.Text
 
 		case <-ctx.Done():
-			err := ctx.Err()
-			if err != nil {
-				log.Errorf("error in done context Tail %v", err)
+			ctxErr := ctx.Err()
+			switch ctxErr {
+			case context.DeadlineExceeded:
+				log.Tracef("Tailer cancelled. Deadline exceeded, %v", ctxErr)
+			case context.Canceled:
+				log.Tracef("Tailer forcibly cancelled, %v", ctxErr)
 			}
-			log.Info("tailer is done")
+			log.Trace("Tailer is done")
 			return
 		}
 	}
@@ -133,11 +137,14 @@ func (t *PatternTailer) Tail(ctx context.Context, data chan<- map[string]string)
 				data <- l
 			}
 		case <-ctx.Done():
-			err := ctx.Err()
-			if err != nil {
-				log.Errorf("error in done context Tail %v", err)
+			ctxErr := ctx.Err()
+			switch ctxErr {
+			case context.DeadlineExceeded:
+				log.Tracef("Tailer cancelled because deadline was exceeded, %v", ctxErr)
+			case context.Canceled:
+				log.Tracef("Tailer forcibly cancelled, %v", ctxErr)
 			}
-			log.Info("tailer is done")
+			log.Error("Tailer is done")
 			return
 		}
 	}

src/plugins/nginx.go

@@ -13,6 +13,8 @@ import (
 	"errors"
 	"fmt"
 	"path/filepath"
+	re "regexp"
+	"sync"
 	"time"
 
 	"github.com/gogo/protobuf/types"
@@ -29,6 +31,7 @@ import (
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/config"
 	"github.com/nginx/agent/v2/src/core/payloads"
+	"github.com/nginx/agent/v2/src/core/tailer"
 	"github.com/nginx/agent/v2/src/extensions/nginx-app-protect/nap"
 )
 
@@ -39,6 +42,9 @@ const (
 
 var (
 	validationTimeout = 15 * time.Second
+	reloadErrorList   = []*re.Regexp{
+		re.MustCompile(`.*Address already in use.*`),
+	}
 )
 
 // Nginx is the metadata of our nginx binary
@@ -383,7 +389,7 @@ func (n *Nginx) writeConfigAndReloadNginx(correlationId string, config *proto.Ng
 	case result := <-n.configApplyStatusChannel:
 		return result
 	case <-time.After(validationTimeout):
-		log.Errorf("Validation of the NGINX config in taking longer than the validationTimeout %s", validationTimeout)
+		log.Warnf("Validation of the NGINX config in taking longer than the validationTimeout %s", validationTimeout)
 		return status
 	}
 }
@@ -466,10 +472,18 @@ func (n *Nginx) completeConfigApply(response *NginxConfigValidationResponse) *pr
 	log.Debug("Enabling file watcher")
 	n.messagePipeline.Process(core.NewMessage(core.FileWatcherEnabled, true))
 
+	var monitoringError error
 	reloadErr := n.nginxBinary.Reload(response.nginxDetails.ProcessId, response.nginxDetails.ProcessPath)
 	if reloadErr != nil {
 		nginxConfigStatusMessage = fmt.Sprintf("Config apply failed (write): %v", reloadErr)
 		log.Errorf(nginxConfigStatusMessage)
+	} else {
+		monitoringError = n.monitor(response.nginxDetails)
+		if (monitoringError != nil) {
+			nginxConfigStatusMessage = fmt.Sprintf("Config apply failed. The following errors were found in the NGINX error logs: %v", monitoringError)
+		} else {
+			log.Info("No errors found in NGINX errors logs after NGINX reload")
+		}
 	}
 
 	nginxReloadEventMeta := NginxReloadResponse{
@@ -481,11 +495,16 @@ func (n *Nginx) completeConfigApply(response *NginxConfigValidationResponse) *pr
 
 	n.messagePipeline.Process(core.NewMessage(core.NginxReloadComplete, nginxReloadEventMeta))
 
+	nginxConfigStatus := proto.NginxConfigStatus_OK
+	if reloadErr != nil || monitoringError != nil {
+		nginxConfigStatus = proto.NginxConfigStatus_ERROR
+	}
+
 	agentActivityStatus := &proto.AgentActivityStatus{
 		Status: &proto.AgentActivityStatus_NginxConfigStatus{
 			NginxConfigStatus: &proto.NginxConfigStatus{
 				CorrelationId: response.correlationId,
-				Status:        proto.NginxConfigStatus_OK,
+				Status:        nginxConfigStatus,
 				Message:       nginxConfigStatusMessage,
 				NginxId:       response.config.GetConfigData().GetNginxId(),
 			},
@@ -513,6 +532,94 @@ func (n *Nginx) completeConfigApply(response *NginxConfigValidationResponse) *pr
 	return status
 }
 
+func (n *Nginx) monitor(details *proto.NginxDetails) error {
+	errorLogs := n.nginxBinary.GetErrorLogs()
+	if len(errorLogs) == 0 {
+		log.Warn("Skipping error log validation for NGINX reload because no error logs have been configured in the NGINX configuration")
+	}
+
+	errorChannel := make(chan string, len(errorLogs))
+	wg := &sync.WaitGroup{}
+
+	for logFile := range errorLogs {
+		wg.Add(1)
+		go n.tailLog(errorChannel, wg, logFile)
+	}
+
+	wg.Add(1)
+	go n.monitorPids(errorChannel, wg)
+
+	wg.Wait()
+	close(errorChannel)
+
+	errorsFound := []string{}
+	for errorLog := range errorChannel {
+		errorsFound = append(errorsFound, errorLog)
+	}
+
+	return errors.New(errorsFound[0])
+}
+
+func (n *Nginx) monitorPids(errorChannel chan string, wg *sync.WaitGroup) {
+	// oldWorkers := utils.GetChildPids(v.Pid)
+	ctx, cncl := context.WithTimeout(context.Background(), 90 * time.Second)
+	defer cncl()
+		t := time.NewTicker(90 * time.Second)
+		for {
+			select {
+			case <-ctx.Done():
+				// log.WithContext(chtx).WithFields(logrus.Fields{
+				// 	"master": v,
+				// 	"pid":    v.Pid,
+				// 	"err":    chtx.Err(),
+				// }).Trace("Context cancelled while waiting for new workers")
+				wg.Done()
+				errorChannel <- "waiting for new workers timed out"
+				return 
+			case <-t.C:
+				// currentWorkers := utils.GetChildPids(v.Pid)
+				// if readyBasedOnWorkers(oldWorkers, currentWorkers) {
+					errorChannel <- ""
+					wg.Done()
+					return
+				//}
+			}
+		}
+}
+
+func (n *Nginx) tailLog(errorChannel chan string, wg *sync.WaitGroup, logFile string) {
+	defer wg.Done()
+
+	t, err := tailer.NewTailer(logFile)
+	if err != nil {
+		log.Errorf("Unable to tail %q: %v", logFile, err)
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	data := make(chan string, 1024)
+	go t.Tail(ctx, data)
+
+	tick := time.NewTicker(n.config.Nginx.ConfigReloadMonitoringPeriod)
+	defer tick.Stop()
+
+	for {
+		select {
+		case d := <-data:
+			for _, errorRegex := range reloadErrorList {
+				if errorRegex.MatchString(d) {
+					errorChannel <- d
+					return
+				}
+			}
+		case <-tick.C:
+			return
+		}
+	}
+}
+
 func (n *Nginx) rollbackConfigApply(response *NginxConfigValidationResponse) {
 	nginxConfigStatusMessage := fmt.Sprintf("Config apply failed (write): %v", response.err.Error())
 	log.Error(nginxConfigStatusMessage)