remove logs/calls apis, async, and most of hybrid (#1458)

* start from the top

remove runner configuration mode

* remove async, logs, calls... hybrid still has one use

* add note

* fix tests

* remove all async verbiage / cruft

* fix test

* remove logs and calls from swagger

* fix rebase

* fix system tests

* remove calls/logs from sql db

adds migration and removes datastore methods

* go mod tidy && go mod vendor

* remove unused env vars

* remove stale server options

Author: rdallman

CONTRIBUTING.md

@@ -11,7 +11,7 @@ The Fn Server is only for the essential parts of the entire Fn ecosystem.
 These include:
 
 - The core API (apps, routes, calls, logs)
-- Executing functions (sync and async)
+- Executing functions (sync)
 - Extension points (callbacks, middleware, API additions)
 
 This does __not__ include:
@@ -86,4 +86,4 @@ Run tests inside a Docker container:
 ```sh
 make docker-test
 
-```
+```

Makefile

@@ -86,21 +86,13 @@ img-mysql:
 img-postgres:
 	/bin/bash -c "source ./helpers.sh && docker_pull_postgres"
 
-.PHONY: img-minio
-img-minio:
-	/bin/bash -c "source ./helpers.sh && docker_pull_minio"
-
 .PHONY: pull-images
-pull-images: img-mysql img-postgres img-minio img-busybox
+pull-images: img-mysql img-postgres img-busybox
 
 .PHONY: test-datastore
 test-datastore:
 	cd api/datastore && go test  ./...
 
-.PHONY: test-log-datastore
-test-log-datastore:
-	cd api/logs && go test  ./...
-
 .PHONY: test-build-arm
 test-build-arm:
 	GOARCH=arm GOARM=5 $(MAKE) build

api/agent/agent.go

@@ -32,33 +32,13 @@ const (
 	pauseTimeout = 5 * time.Second // docker pause/unpause
 )
 
-// TODO we should prob store async calls in db immediately since we're returning id (will 404 until post-execution)
-// TODO async calls need to add route.Headers as well
-// TODO handle timeouts / no response in sync & async (sync is json+503 atm, not 504, async is empty log+status)
-// see also: server/runner.go wrapping the response writer there, but need to handle async too (push down?)
-// TODO storing logs / call can push call over the timeout
-// TODO async is still broken, but way less so. we need to modify mq semantics
-// to be much more robust. now we're at least running it if we delete the msg,
-// but we may never store info about that execution so still broked (if fn
-// dies). need coordination w/ db.
-// TODO if async would store requests (or interchange format) it would be slick, but
-// if we're going to store full calls in db maybe we should only queue pointers to ids?
-// TODO examine cases where hot can't start a container and the user would never see an error
-// about why that may be so (say, whatever it is takes longer than the timeout, e.g.)
-// TODO if an image is not found or similar issues in getting a slot, then async should probably
-// mark the call as errored rather than forever trying & failing to run it
-// TODO it would be really nice if we made the ramToken wrap the driver cookie (less brittle,
-// if those leak the container leaks too...) -- not the allocation, but the token.Close and cookie.Close
-// TODO if machine is out of ram, just timeout immediately / wait for hot slot? (discuss policy)
-
 // Agent exposes an api to create calls from various parameters and then submit
 // those calls, it also exposes a 'safe' shutdown mechanism via its Close method.
 // Agent has a few roles:
 //	* manage the memory pool for a given server
 //	* manage the container lifecycle for calls
 //	* execute calls against containers
 //	* invoke Start and End for each call appropriately
-//	* check the mq for any async calls, and submit them
 //
 // Overview:
 // Upon submission of a call, Agent will start the call's timeout timer
@@ -96,7 +76,6 @@ type Agent interface {
 
 type agent struct {
 	cfg           Config
-	da            CallHandler
 	callListeners []fnext.CallListener
 
 	driver drivers.Driver
@@ -128,7 +107,7 @@ type Option func(*agent) error
 const RegistryToken = "FN_REGISTRY_TOKEN"
 
 // New creates an Agent that executes functions locally as Docker containers.
-func New(da CallHandler, options ...Option) Agent {
+func New(options ...Option) Agent {
 
 	cfg, err := NewConfig()
 	if err != nil {
@@ -140,7 +119,6 @@ func New(da CallHandler, options ...Option) Agent {
 	}
 
 	a.shutWg = common.NewWaitGroup()
-	a.da = da
 	a.slotMgr = NewSlotQueueMgr()
 	a.evictor = NewEvictor()
 
@@ -175,19 +153,6 @@ func (a *agent) addStartup(sup func()) {
 
 }
 
-// WithAsync Enables Async  operations on the agent
-func WithAsync(dqda DequeueDataAccess) Option {
-	return func(a *agent) error {
-		a.addStartup(func() {
-			if !a.shutWg.AddSession(1) {
-				logrus.Fatal("cannot start agent, unable to add session")
-			}
-			go a.asyncDequeue(dqda) // safe shutdown can nanny this fine
-		})
-		return nil
-	}
-}
-
 // WithConfig sets the agent config to the provided config
 func WithConfig(cfg *Config) Option {
 	return func(a *agent) error {
@@ -353,18 +318,6 @@ func (a *agent) handleCallEnd(ctx context.Context, call *call, slot Slot, err er
 // for other containers to become idle or it may wait for resources to become
 // available to launch a new container.
 func (a *agent) getSlot(ctx context.Context, call *call) (Slot, error) {
-	if call.Type == models.TypeAsync {
-		// *) for async, slot deadline is also call.Timeout. This is because we would like to
-		// allocate enough time for docker-pull, slot-wait, docker-start, etc.
-		// and also make sure we have call.Timeout inside the container. Total time
-		// to run an async becomes 2 * call.Timeout.
-		// *) for sync, there's no slot deadline, the timeout is controlled by http-client
-		// context (or runner gRPC context)
-		tmp, cancel := context.WithTimeout(ctx, time.Duration(call.Timeout)*time.Second)
-		ctx = tmp
-		defer cancel()
-	}
-
 	ctx, span := trace.StartSpan(ctx, "agent_get_slot")
 	defer span.End()
 
@@ -714,8 +667,7 @@ func (s *hotSlot) dispatch(ctx context.Context, call *call) error {
 func (s *hotSlot) writeResp(ctx context.Context, max uint64, resp *http.Response, w io.Writer) error {
 	rw, ok := w.(http.ResponseWriter)
 	if !ok {
-		// WARNING: this bypasses container contract translation. Assuming this is
-		// async mode, where we are storing response in call.stderr.
+		// TODO(reed): this is strange, we should just enforce the response writer type?
 		w = common.NewClampWriter(w, max, models.ErrFunctionResponseTooBig)
 		return resp.Write(w)
 	}

api/agent/agent_evict_test.go

@@ -14,15 +14,12 @@ import (
 	"github.com/fnproject/fn/api/agent/drivers"
 	_ "github.com/fnproject/fn/api/agent/drivers/docker"
 	"github.com/fnproject/fn/api/id"
-	"github.com/fnproject/fn/api/logs"
 	"github.com/fnproject/fn/api/models"
-	"github.com/fnproject/fn/api/mqs"
 )
 
 // create a simple non-blocking agent. Non-blocking does not queue, so it's
 // easier to test and see if evictions took place.
 func getAgentWithDriver() (Agent, drivers.Driver, error) {
-	ls := logs.NewMock()
 	cfg, err := NewConfig()
 	if err != nil {
 		return nil, nil, err
@@ -40,7 +37,7 @@ func getAgentWithDriver() (Agent, drivers.Driver, error) {
 		return nil, nil, err
 	}
 
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg), WithDockerDriver(drv))
+	a := New(WithConfig(cfg), WithDockerDriver(drv))
 	return a, drv, nil
 }
 

api/agent/agent_test.go

@@ -23,9 +23,7 @@ import (
 	_ "github.com/fnproject/fn/api/agent/drivers/docker"
 	"github.com/fnproject/fn/api/common"
 	"github.com/fnproject/fn/api/id"
-	"github.com/fnproject/fn/api/logs"
 	"github.com/fnproject/fn/api/models"
-	"github.com/fnproject/fn/api/mqs"
 
 	"github.com/sirupsen/logrus"
 )
@@ -96,9 +94,7 @@ func TestCallConfigurationRequest(t *testing.T) {
 		},
 	}
 
-	ls := logs.NewMock()
-
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	w := httptest.NewRecorder()
@@ -140,9 +136,6 @@ func TestCallConfigurationRequest(t *testing.T) {
 	if model.Type != "sync" {
 		t.Fatal("fn type mismatch", model.Type)
 	}
-	if model.Priority == nil {
-		t.Fatal("GetCall should make priority non-nil so that async works because for whatever reason some clowns plumbed it all over the mqs even though the user can't specify it gg")
-	}
 	if model.Timeout != timeout {
 		t.Fatal("timeout mismatch", model.Timeout, timeout)
 	}
@@ -222,10 +215,7 @@ func TestCallConfigurationModel(t *testing.T) {
 		Method:      method,
 	}
 
-	// FromModel doesn't need a datastore, for now...
-	ls := logs.NewMock()
-
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	callI, err := a.GetCall(FromModel(cm))
@@ -259,10 +249,7 @@ func TestGetCallFromModelRoundTripACall(t *testing.T) {
 		Payload: payload,
 	}
 
-	// FromModel doesn't need a datastore, for now...
-	ls := logs.NewMock()
-
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	callI, err := a.GetCall(FromModel(cm))
@@ -395,10 +382,7 @@ func TestSubmitError(t *testing.T) {
 		Method:      method,
 	}
 
-	// FromModel doesn't need a datastore, for now...
-	ls := logs.NewMock()
-
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	var wg sync.WaitGroup
@@ -453,10 +437,9 @@ func TestHungFDK(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
 	cfg, err := NewConfig()
 	cfg.HotStartTimeout = time.Duration(3) * time.Second
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	req, err := http.NewRequest("GET", url, &dummyReader{Reader: strings.NewReader(`{}`)})
@@ -507,10 +490,9 @@ func TestDockerPullHungRepo(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
 	cfg, err := NewConfig()
 	cfg.HotPullTimeout = time.Duration(5) * time.Second
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	req, err := http.NewRequest("GET", url, &dummyReader{Reader: strings.NewReader(`{}`)})
@@ -560,10 +542,9 @@ func TestDockerPullUnAuthorizedRepo(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
 	cfg, err := NewConfig()
 	cfg.HotPullTimeout = time.Duration(5) * time.Second
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	req, err := http.NewRequest("GET", url, &dummyReader{Reader: strings.NewReader(`{}`)})
@@ -608,9 +589,8 @@ func TestDockerPullBadRepo(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
 	cfg, err := NewConfig()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	req, err := http.NewRequest("GET", url, &dummyReader{Reader: strings.NewReader(`{}`)})
@@ -634,8 +614,6 @@ func TestDockerPullBadRepo(t *testing.T) {
 }
 
 func TestHTTPWithoutContentLengthWorks(t *testing.T) {
-	// TODO it may be a good idea to mock out the http server and use a real
-	// response writer with sync, and also test that this works with async + log
 	app := &models.App{ID: "app_id"}
 	fn := &models.Fn{
 		ID:    "fn_id",
@@ -649,8 +627,7 @@ func TestHTTPWithoutContentLengthWorks(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	bodOne := `{"echoContent":"yodawg"}`
@@ -709,8 +686,7 @@ func TestGetCallReturnsResourceImpossibility(t *testing.T) {
 		Memory:      math.MaxUint64,
 	}
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	_, err := a.GetCall(FromModel(call))
@@ -738,8 +714,7 @@ func TestTmpFsRW(t *testing.T) {
 
 	url := "http://127.0.0.1:8080/invoke/" + fn.ID
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	// Here we tell fn-test-utils to read file /proc/mounts and create a /tmp/salsa of 4MB
@@ -833,8 +808,7 @@ func TestTmpFsSize(t *testing.T) {
 
 	cfg.MaxTmpFsInodes = 1025
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	// Here we tell fn-test-utils to read file /proc/mounts and create a /tmp/salsa of 4MB
@@ -988,8 +962,7 @@ func TestPipesAreClear(t *testing.T) {
 		},
 	}
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	// test read this body after 5s (after call times out) and make sure we don't get yodawg
@@ -1132,8 +1105,7 @@ func TestCallsDontInterlace(t *testing.T) {
 		},
 	}
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	bodOne := `{"echoContent":"yodawg"}`
@@ -1233,8 +1205,7 @@ func TestNBIOResourceTracker(t *testing.T) {
 	cfg.MaxTotalMemory = 280 * 1024 * 1024
 	cfg.HotPoll = 20 * time.Millisecond
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)), WithConfig(cfg))
+	a := New(WithConfig(cfg))
 	defer checkClose(t, a)
 
 	reqCount := 20
@@ -1300,8 +1271,7 @@ func TestDockerAuthExtn(t *testing.T) {
 		t.Fatalf("bad config %+v", cfg)
 	}
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	callIf, err := a.GetCall(FromModel(modelCall))
@@ -1421,8 +1391,7 @@ func TestContainerDisableIO(t *testing.T) {
 		t.Fatalf("bad config %+v", cfg)
 	}
 
-	ls := logs.NewMock()
-	a := New(NewDirectCallDataAccess(ls, new(mqs.Mock)))
+	a := New()
 	defer checkClose(t, a)
 
 	// NOTE: right now we disable stdin by default so this test should pass.