feat: connect admission controllers

admissionctrl/controller.go

@@ -24,12 +24,21 @@ type JobValidator interface {
 	AdmissionController
 	Validate(*structs.Job) (warnings []error, err error)
 }
+
 type JobHandler struct {
 	mutators   []JobMutator
 	validators []JobValidator
 	logger     hclog.Logger
 }
 
+func NewJobHandler(mutators []JobMutator, validators []JobValidator, logger hclog.Logger) *JobHandler {
+	return &JobHandler{
+		mutators:   mutators,
+		validators: validators,
+		logger:     logger,
+	}
+}
+
 func (j *JobHandler) ApplyAdmissionControllers(job *structs.Job) (out *structs.Job, warnings []error, err error) {
 	// Mutators run first before validators, so validators view the final rendered job.
 	// So, mutators must handle invalid jobs.

main.go

@@ -1,11 +1,19 @@
 package main
 
 import (
-	"fmt"
+	"bytes"
+	"encoding/json"
+	"io"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"os"
 	"regexp"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/mxab/nacp/admissionctrl"
+	"github.com/mxab/nacp/admissionctrl/mutator"
 )
 
 type NomadServer struct {
@@ -17,7 +25,7 @@ type Config struct {
 	nomad NomadServer
 }
 
-func NewServer(c Config) *httputil.ReverseProxy {
+func NewServer(c Config, appLogger hclog.Logger) *httputil.ReverseProxy {
 
 	// create a reverse proxy that catches "/v1/jobs" post calls
 	// and forwards them to the jobs service
@@ -26,31 +34,74 @@ func NewServer(c Config) *httputil.ReverseProxy {
 	if err != nil {
 		panic(err)
 	}
+
+	handler := admissionctrl.NewJobHandler(
+
+		[]admissionctrl.JobMutator{&mutator.HelloMutator{}},
+		[]admissionctrl.JobValidator{},
+		appLogger.Named("handler"),
+	)
+
 	proxy := httputil.NewSingleHostReverseProxy(backend)
 	originalDirector := proxy.Director
+
 	proxy.Director = func(r *http.Request) {
 		originalDirector(r)
 
 		// only check for the header if it is a POST request and path is /v1/jobs
-		if r.Method == "POST" && r.URL.Path == "/v1/jobs" {
-			r.Header.Set("NACP", "CREATE")
-		}
+		if isCreate(r) || isUpdate(r) {
+
+			jobRegisterRequest := &structs.JobRegisterRequest{}
+			if err := json.NewDecoder(r.Body).Decode(jobRegisterRequest); err != nil {
+				//TODO: configure if we want to prevent the request from being forwarded
+				appLogger.Info("Failed decoding job, skipping admission controller", "error", err)
+				return
+			}
+			job, warnings, err := handler.ApplyAdmissionControllers(jobRegisterRequest.Job)
+			if err != nil {
+				//TODO: configure if we want to prevent the request from being forwarded
+
+				appLogger.Warn("Failed to apply admission controllers, skipping", "error", err)
+				return
+			}
+			if len(warnings) > 0 {
+				appLogger.Warn("Warnings applying admission controllers", "warnings", warnings)
+			}
+			jobRegisterRequest.Job = job
+
+			data, err := json.Marshal(jobRegisterRequest)
+
+			if err != nil {
+				//TODO: configure if we want to prevent the request from being forwarded
+				appLogger.Warn("Error marshalling job", "error", err)
+				return
+			}
+			r.ContentLength = int64(len(data))
+			r.Body = io.NopCloser(bytes.NewBuffer(data))
 
-		// only check for the header if it is a POST request
-		// and path matches a regex that starts with /v1/job/ followed by a reg job name (e.g. /v1/job/some-job)
-		if r.Method == "POST" && regexp.MustCompile("^/v1/job/.*").MatchString(r.URL.Path) {
-			r.Header.Set("NACP", "UPDATE")
 		}
 
 	}
 	return proxy
 }
+func isCreate(r *http.Request) bool {
+	return r.Method == "POST" && r.URL.Path == "/v1/jobs"
+}
+func isUpdate(r *http.Request) bool {
+	return r.Method == "POST" && regexp.MustCompile("^/v1/job/.*").MatchString(r.URL.Path)
+}
 
 // https://www.codedodle.com/go-reverse-proxy-example.html
 func main() {
 
-	fmt.Println("Hello")
-	// create a reverse proxy that catches "/v1/jobs" post calls
+	appLogger := hclog.New(&hclog.LoggerOptions{
+		Name:   "nacp",
+		Level:  hclog.LevelFromString("DEBUG"),
+		Output: os.Stdout,
+	})
+
+	appLogger.Info("Starting Nomad Admission Control Proxy")
+
 	// and forwards them to the jobs service
 	// create a new reverse proxy
 
@@ -61,6 +112,6 @@ func main() {
 			Address: "http://localhost:4646",
 		},
 	}
-	NewServer(c)
+	NewServer(c, appLogger)
 
 }

main_test.go

@@ -1,12 +1,17 @@
 package main
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
-	"strings"
 	"testing"
 
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/mxab/nacp/testutil"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -19,18 +24,29 @@ func TestProxyTableDriven(t *testing.T) {
 		method string
 		want   string
 	}
+	wantJob := testutil.ReadJob(t)
+	wantJob.Meta = map[string]string{
+		"hello": "world",
+	}
+	register := &structs.JobRegisterRequest{
+		Job: wantJob,
+	}
+	wantRegisterData, err := json.Marshal(register)
+	if err != nil {
+		t.Fatal(err)
+	}
 	tests := []test{
 		{
 			name:   "create job",
 			path:   "/v1/jobs",
 			method: "POST",
-			want:   "CREATE",
+			want:   string(wantRegisterData),
 		},
 		{
 			name:   "update job",
 			path:   "/v1/job/some-job",
 			method: "POST",
-			want:   "UPDATE",
+			want:   string(wantRegisterData),
 		},
 	}
 
@@ -41,8 +57,12 @@ func TestProxyTableDriven(t *testing.T) {
 
 				assert.Equal(t, req.Method, tc.method, "Ensure method is set to POST")
 				assert.Equal(t, req.URL.Path, tc.path, "Ensure path is set")
-				assert.Equal(t, req.Header.Get("NACP"), tc.want, "Ensure NACP header is set to CREATE")
-
+				jsonData, err := io.ReadAll(req.Body)
+				if err != nil {
+					t.Fatal(err)
+				}
+				json := string(jsonData)
+				assert.JSONEq(t, tc.want, json, "Body matches")
 				_, _ = rw.Write([]byte(`OK`))
 			}))
 			// Close the server when test finishes
@@ -56,16 +76,23 @@ func TestProxyTableDriven(t *testing.T) {
 					Address: nomadDummy.URL,
 				},
 			}
-			proxy := NewServer(c)
+			proxy := NewServer(c, hclog.NewNullLogger())
 
 			//http.Handle("/", proxy)
 
 			proxyServer := httptest.NewServer(proxy)
 			defer proxyServer.Close()
 
-			res, err := http.Post(fmt.Sprintf("%s%s", proxyServer.URL, tc.path), "application/json", strings.NewReader(`
-			{}
-			`))
+			jobRequest := structs.JobRegisterRequest{
+				Job: testutil.ReadJob(t),
+			}
+
+			buffer := new(bytes.Buffer)
+			err := json.NewEncoder(buffer).Encode(jobRequest)
+			if err != nil {
+				t.Fatal(err)
+			}
+			res, err := http.Post(fmt.Sprintf("%s%s", proxyServer.URL, tc.path), "application/json", buffer)
 			assert.NoError(t, err, "No http call error")
 			assert.Equal(t, res.StatusCode, 200, "OK response is expected")
 		})