Merge 8767c87 into a6eb7a2

Dockerfile

@@ -31,4 +31,8 @@ COPY --from=builder /build/pg_timetable /
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 
 # Command to run the executable
-ENTRYPOINT ["/pg_timetable"]
+ENTRYPOINT ["/pg_timetable"]
+
+ENV PGTT_RESTPORT 8088
+# Exposing port 8088 (you should specify your value!)
+EXPOSE 8088

internal/api/status.go

@@ -0,0 +1,52 @@
+package api
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/cybertec-postgresql/pg_timetable/internal/config"
+	"github.com/cybertec-postgresql/pg_timetable/internal/log"
+)
+
+// StatusIface is a common interface describing the cirrent status of a connection
+type StatusIface interface {
+	IsReady() bool
+}
+
+type RestApiServer struct {
+	StatusReporter StatusIface
+	l              log.LoggerIface
+	*http.Server
+}
+
+func Init(opts config.RestApiOpts, logger log.LoggerIface) *RestApiServer {
+	s := &RestApiServer{
+		nil,
+		logger,
+		&http.Server{
+			Addr:           fmt.Sprintf(":%d", opts.Port),
+			ReadTimeout:    10 * time.Second,
+			WriteTimeout:   10 * time.Second,
+			MaxHeaderBytes: 1 << 20,
+		},
+	}
+	http.HandleFunc("/liveness", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK) // i'm serving hence I'm alive
+	})
+	http.HandleFunc("/readiness", s.readinessHandler)
+	if opts.Port != 0 {
+		logger.WithField("port", opts.Port).Info("Starting REST API server...")
+		go func() { logger.Error(s.ListenAndServe()) }()
+	}
+	return s
+}
+
+func (Server *RestApiServer) readinessHandler(w http.ResponseWriter, r *http.Request) {
+	Server.l.Debug("Received /readiness REST API request")
+	if Server.StatusReporter == nil || !Server.StatusReporter.IsReady() {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}

internal/config/cmdparser.go

@@ -43,6 +43,11 @@ type ResourceOpts struct {
 	TaskTimeout     int `long:"task-timeout" mapstructure:"task-timeout" description:"Abort any task within a chain that takes more than the specified number of milliseconds"`
 }
 
+// RestApiOpts fot internal web server impleenting REST API
+type RestApiOpts struct {
+	Port int `long:"rest-port" mapstructure:"rest-port" description:"HTTP port for REST API" env:"PGTT_RESTPORT" default:"0"`
+}
+
 // CmdOptions holds command line options passed
 type CmdOptions struct {
 	ClientName     string         `short:"c" long:"clientname" description:"Unique name for application instance" env:"PGTT_CLIENTNAME"`
@@ -51,6 +56,7 @@ type CmdOptions struct {
 	Logging        LoggingOpts    `group:"Logging" mapstructure:"Logging"`
 	Start          StartOpts      `group:"Start" mapstructure:"Start"`
 	Resource       ResourceOpts   `group:"Resource" mapstructure:"Resource"`
+	RestApi        RestApiOpts    `group:"HTTP" mapstructure:"HTTP"`
 	NoProgramTasks bool           `long:"no-program-tasks" mapstructure:"no-program-tasks" description:"Disable executing of PROGRAM tasks" env:"PGTT_NOPROGRAMTASKS"`
 	NoHelpMessage  bool           `long:"no-help" mapstructure:"no-help" hidden:"system use"`
 }

internal/pgengine/pgengine_test.go

@@ -237,7 +237,7 @@ func TestSamplesScripts(t *testing.T) {
 		assert.NoError(t, pge.ExecuteCustomScripts(ctx, "../../samples/"+f.Name()),
 			"Sample query failed: ", f.Name())
 		// either context should be cancelled or 'shutdown.sql' will terminate the session
-		assert.True(t, scheduler.New(pge, l).Run(ctx) > scheduler.ConnectionDropppedStatus)
+		assert.True(t, scheduler.New(pge, l).Run(ctx) > scheduler.RunningStatus)
 		_, err = pge.ConfigDb.Exec(context.Background(),
 			"TRUNCATE timetable.task, timetable.chain CASCADE")
 		assert.NoError(t, err)

internal/scheduler/scheduler.go

@@ -20,8 +20,8 @@ const minChannelCapacity = 1024
 type RunStatus int
 
 const (
-	// ConnectionDropppedStatus specifies the connection has been dropped
-	ConnectionDropppedStatus RunStatus = iota
+	// RunningStatus specifies the connection has been dropped
+	RunningStatus RunStatus = iota
 	// ContextCancelledStatus specifies the context has been cancelled probably due to timeout
 	ContextCancelledStatus
 	// Shutdown specifies proper termination of the session
@@ -46,6 +46,7 @@ type Scheduler struct {
 	intervalChainsChan chan IntervalChain
 	intervalChainMutex sync.Mutex
 	shutdown           chan struct{} // closed when shutdown is called
+	status             RunStatus
 }
 
 func max(x, y int) int {
@@ -65,6 +66,7 @@ func New(pge *pgengine.PgEngine, logger log.LoggerIface) *Scheduler {
 		activeChains:       make(map[int]func()), //holds cancel() functions to stop chains
 		intervalChains:     make(map[int]IntervalChain),
 		shutdown:           make(chan struct{}),
+		status:             RunningStatus,
 	}
 }
 
@@ -78,6 +80,11 @@ func (sch *Scheduler) Config() config.CmdOptions {
 	return sch.pgengine.CmdOptions
 }
 
+// IsReady returns True if the scheduler is in the main loop processing chains
+func (sch *Scheduler) IsReady() bool {
+	return sch.status == RunningStatus
+}
+
 // Run executes jobs. Returns RunStatus why it terminated.
 // There are only two possibilities: dropped connection and cancelled context.
 func (sch *Scheduler) Run(ctx context.Context) RunStatus {
@@ -122,10 +129,14 @@ func (sch *Scheduler) Run(ctx context.Context) RunStatus {
 		case <-time.After(refetchTimeout * time.Second):
 			// pass
 		case <-ctx.Done():
-			return ContextCancelledStatus
+			sch.status = ContextCancelledStatus
 		case <-sch.shutdown:
+			sch.status = ShutdownStatus
 			sch.terminateChains()
-			return ShutdownStatus
+		}
+
+		if sch.status != RunningStatus {
+			return sch.status
 		}
 	}
 }

main.go

@@ -7,6 +7,7 @@ import (
 	"os/signal"
 	"syscall"
 
+	"github.com/cybertec-postgresql/pg_timetable/internal/api"
 	"github.com/cybertec-postgresql/pg_timetable/internal/config"
 	"github.com/cybertec-postgresql/pg_timetable/internal/log"
 	"github.com/cybertec-postgresql/pg_timetable/internal/pgengine"
@@ -62,6 +63,9 @@ func main() {
 		return
 	}
 	defer pge.Finalize()
+
+	apiserver := api.Init(cmdOpts.RestApi, logger)
+
 	if cmdOpts.Start.Upgrade {
 		if err := pge.MigrateDb(ctx); err != nil {
 			logger.WithError(err).Error("Upgrade failed")
@@ -84,7 +88,9 @@ func main() {
 		return
 	}
 	sch := scheduler.New(pge, logger)
-	for sch.Run(ctx) == scheduler.ConnectionDropppedStatus {
+	apiserver.StatusReporter = sch
+
+	for sch.Run(ctx) == scheduler.RunningStatus {
 		pge.ReconnectAndFixLeftovers(ctx)
 	}
 }