S3 loggyloo (#511)

* add minio-go dep, update deps

* add minio s3 client

minio has an s3 compatible api and is an open source project and, notably, is
not amazon, so it seems best to use their client (fwiw the aws-sdk-go is a
giant hair ball of things we don't need, too). it was pretty easy and seems
to work, so rolling with it. also, minio is a totally feasible option for fn
installs in prod / for demos / for local.

* adds 's3' package for s3 compatible log storage api, for use with storing
logs from calls and retrieving them.
* removes DELETE /v1/apps/:app/calls/:call/log endpoint
* removes internal log deletion api
* changes the GetLog API to use an io.Reader, which is a backwards step atm
due to the json api for logs, I have another branch lined up to make a plain
text log API and this will be much more efficient (also want to gzip)
* hooked up minio to the test suite and fixed up the test suite
* add how to run minio docs and point fn at it docs

some notes: notably we aren't cleaning up these logs. there is a ticket
already to make a Mr. Clean who wakes up periodically and nukes old stuff, so
am punting any api design around some kind of TTL deletion of logs. there are
a lot of options really for Mr. Clean, we can notably defer to him when apps
are deleted, too, so that app deletion is fast and then Mr. Clean will just
clean them up later (seems like a good option).

have not tested against BMC object store, which has an s3 compatible API. but
in theory it 'just works' (the reason for doing this). in any event, that's
part of the service land to figure out.

closes #481
closes #473

* add log not found error to minio land

api/agent/agent.go

@@ -119,6 +119,7 @@ type agent struct {
 	// TODO maybe these should be on GetCall? idk. was getting bloated.
 	mq            models.MessageQueue
 	ds            models.Datastore
+	ls            models.LogStore
 	callListeners []extensions.CallListener
 
 	driver drivers.Driver
@@ -139,12 +140,13 @@ type agent struct {
 	promHandler http.Handler
 }
 
-func New(ds models.Datastore, mq models.MessageQueue) Agent {
+func New(ds models.Datastore, ls models.LogStore, mq models.MessageQueue) Agent {
 	// TODO: Create drivers.New(runnerConfig)
 	driver := docker.NewDocker(drivers.Config{})
 
 	a := &agent{
 		ds:          ds,
+		ls:          ls,
 		mq:          mq,
 		driver:      driver,
 		hot:         make(map[string]chan slot),

api/agent/agent_test.go

@@ -49,7 +49,7 @@ func TestCallConfigurationRequest(t *testing.T) {
 		}, nil,
 	)
 
-	a := New(ds, new(mqs.Mock))
+	a := New(ds, ds, new(mqs.Mock))
 	defer a.Close()
 
 	w := httptest.NewRecorder()
@@ -237,7 +237,7 @@ func TestCallConfigurationModel(t *testing.T) {
 	// FromModel doesn't need a datastore, for now...
 	ds := datastore.NewMockInit(nil, nil, nil)
 
-	a := New(ds, new(mqs.Mock))
+	a := New(ds, ds, new(mqs.Mock))
 	defer a.Close()
 
 	callI, err := a.GetCall(FromModel(cm))

api/agent/call.go

@@ -246,8 +246,8 @@ func (a *agent) GetCall(opts ...CallOpt) (Call, error) {
 		return nil, errors.New("no model or request provided for call")
 	}
 
-	// TODO add log store interface (yagni?)
 	c.ds = a.ds
+	c.ls = a.ls
 	c.mq = a.mq
 
 	ctx, _ := common.LoggerWithFields(c.req.Context(),
@@ -270,6 +270,7 @@ type call struct {
 	*models.Call
 
 	ds     models.Datastore
+	ls     models.LogStore
 	mq     models.MessageQueue
 	w      io.Writer
 	req    *http.Request
@@ -353,7 +354,7 @@ func (c *call) End(ctx context.Context, errIn error, t callTrigger) error {
 		// note: Not returning err here since the job could have already finished successfully.
 	}
 
-	if err := c.ds.InsertLog(ctx, c.AppName, c.ID, c.stderr); err != nil {
+	if err := c.ls.InsertLog(ctx, c.AppName, c.ID, c.stderr); err != nil {
 		common.Logger(ctx).WithError(err).Error("error uploading log")
 		// note: Not returning err here since the job could have already finished successfully.
 	}

api/datastore/internal/datastoreutil/metrics.go

@@ -101,17 +101,11 @@ func (m *metricds) InsertLog(ctx context.Context, appName, callID string, callLo
 	return m.ds.InsertLog(ctx, appName, callID, callLog)
 }
 
-func (m *metricds) GetLog(ctx context.Context, appName, callID string) (*models.CallLog, error) {
+func (m *metricds) GetLog(ctx context.Context, appName, callID string) (io.Reader, error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "ds_get_log")
 	defer span.Finish()
 	return m.ds.GetLog(ctx, appName, callID)
 }
 
-func (m *metricds) DeleteLog(ctx context.Context, appName, callID string) error {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "ds_delete_log")
-	defer span.Finish()
-	return m.ds.DeleteLog(ctx, appName, callID)
-}
-
 // instant & no context ;)
 func (m *metricds) GetDatabase() *sqlx.DB { return m.ds.GetDatabase() }

api/datastore/internal/datastoreutil/validator.go

@@ -130,10 +130,6 @@ func (v *validator) GetCall(ctx context.Context, appName, callID string) (*model
 	return v.Datastore.GetCall(ctx, appName, callID)
 }
 
-func (v *validator) DeleteLog(ctx context.Context, appName, callID string) error {
-	return v.Datastore.DeleteLog(ctx, appName, callID)
-}
-
 // GetDatabase returns the underlying sqlx database implementation
 func (v *validator) GetDatabase() *sqlx.DB {
 	return v.Datastore.GetDatabase()

api/datastore/sql/sql.go

@@ -660,7 +660,7 @@ func (ds *sqlStore) InsertLog(ctx context.Context, appName, callID string, logR
 	return err
 }
 
-func (ds *sqlStore) GetLog(ctx context.Context, appName, callID string) (*models.CallLog, error) {
+func (ds *sqlStore) GetLog(ctx context.Context, appName, callID string) (io.Reader, error) {
 	query := ds.db.Rebind(`SELECT log FROM logs WHERE id=? AND app_name=?`)
 	row := ds.db.QueryRowContext(ctx, query, callID, appName)
 
@@ -673,17 +673,7 @@ func (ds *sqlStore) GetLog(ctx context.Context, appName, callID string) (*models
 		return nil, err
 	}
 
-	return &models.CallLog{
-		CallID:  callID,
-		Log:     log,
-		AppName: appName,
-	}, nil
-}
-
-func (ds *sqlStore) DeleteLog(ctx context.Context, appName, callID string) error {
-	query := ds.db.Rebind(`DELETE FROM logs WHERE id=? AND app_name=?`)
-	_, err := ds.db.ExecContext(ctx, query, callID, appName)
-	return err
+	return strings.NewReader(log), nil
 }
 
 func buildFilterRouteQuery(filter *models.RouteFilter) (string, []interface{}) {

api/logs/log.go

@@ -5,6 +5,7 @@ import (
 	"net/url"
 
 	"github.com/fnproject/fn/api/datastore/sql"
+	"github.com/fnproject/fn/api/logs/s3"
 	"github.com/fnproject/fn/api/models"
 	"github.com/sirupsen/logrus"
 )
@@ -18,6 +19,8 @@ func New(dbURL string) (models.LogStore, error) {
 	switch u.Scheme {
 	case "sqlite3", "postgres", "mysql":
 		return sql.New(u)
+	case "s3":
+		return s3.New(u)
 	default:
 		return nil, fmt.Errorf("db type not supported %v", u.Scheme)
 	}

api/logs/log_test.go

@@ -22,5 +22,5 @@ func TestDatastore(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create sqlite3 datastore: %v", err)
 	}
-	logTesting.Test(t, ds, ds)
+	logTesting.Test(t, ds)
 }

api/logs/mock.go

@@ -1,52 +1,30 @@
 package logs
 
 import (
-	"bytes"
 	"context"
 	"io"
 
 	"github.com/fnproject/fn/api/models"
-	"github.com/pkg/errors"
 )
 
 type mock struct {
-	Logs map[string]*models.CallLog
-	ds   models.Datastore
+	Logs map[string]io.Reader
 }
 
 func NewMock() models.LogStore {
-	return NewMockInit(nil)
-}
-
-func NewMockInit(logs map[string]*models.CallLog) models.LogStore {
-	if logs == nil {
-		logs = map[string]*models.CallLog{}
-	}
-	fnl := &mock{logs, nil}
-	return fnl
-}
-
-func (m *mock) SetDatastore(ctx context.Context, ds models.Datastore) {
-	m.ds = ds
+	return &mock{make(map[string]io.Reader)}
 }
 
 func (m *mock) InsertLog(ctx context.Context, appName, callID string, callLog io.Reader) error {
-	var b bytes.Buffer
-	io.Copy(&b, callLog)
-	m.Logs[callID] = &models.CallLog{CallID: callID, Log: b.String()}
+	m.Logs[callID] = callLog
 	return nil
 }
 
-func (m *mock) GetLog(ctx context.Context, appName, callID string) (*models.CallLog, error) {
+func (m *mock) GetLog(ctx context.Context, appName, callID string) (io.Reader, error) {
 	logEntry := m.Logs[callID]
 	if logEntry == nil {
-		return nil, errors.New("Call log not found")
+		return nil, models.ErrCallLogNotFound
 	}
 
-	return m.Logs[callID], nil
-}
-
-func (m *mock) DeleteLog(ctx context.Context, appName, callID string) error {
-	delete(m.Logs, callID)
-	return nil
+	return logEntry, nil
 }

api/logs/s3/s3.go

@@ -0,0 +1,109 @@
+// package s3 implements an s3 api compatible log store
+package s3
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"io"
+	"net/url"
+	"strings"
+
+	"github.com/fnproject/fn/api/models"
+	"github.com/minio/minio-go"
+	"github.com/sirupsen/logrus"
+)
+
+// TODO we should encrypt these, user will have to supply a key though (or all
+// OSS users logs will be encrypted with same key unless they change it which
+// just seems mean...)
+
+// TODO do we need to use the v2 API? can't find BMC object store docs :/
+
+const (
+	contentType = "text/plain"
+)
+
+type store struct {
+	client *minio.Client
+	bucket string
+}
+
+// s3://access_key_id:secret_access_key@host/location/bucket_name?ssl=true
+func New(u *url.URL) (models.LogStore, error) {
+	endpoint := u.Host
+
+	var accessKeyID, secretAccessKey string
+	if u.User != nil {
+		accessKeyID = u.User.Username()
+		secretAccessKey, _ = u.User.Password()
+	}
+	useSSL := u.Query().Get("ssl") == "true"
+
+	strs := strings.SplitN(u.Path, "/", 3)
+	if len(strs) < 3 {
+		return nil, errors.New("must provide bucket name and region in path of s3 api url. e.g. s3://s3.com/us-east-1/my_bucket")
+	}
+	location := strs[1]
+	bucketName := strs[2]
+	if location == "" {
+		return nil, errors.New("must provide non-empty location in path of s3 api url. e.g. s3://s3.com/us-east-1/my_bucket")
+	} else if bucketName == "" {
+		return nil, errors.New("must provide non-empty bucket name in path of s3 api url. e.g. s3://s3.com/us-east-1/my_bucket")
+	}
+
+	logrus.WithFields(logrus.Fields{"bucketName": bucketName, "location": location, "endpoint": endpoint, "access_key_id": accessKeyID, "useSSL": useSSL}).Info("checking / creating s3 bucket")
+
+	client, err := minio.New(endpoint, accessKeyID, secretAccessKey, useSSL)
+	if err != nil {
+		return nil, err
+	}
+
+	// ensure the bucket exists, creating if it does not
+	err = client.MakeBucket(bucketName, location)
+	if errMake := err; err != nil {
+		// Check to see if we already own this bucket (which happens if you run this twice)
+		exists, err := client.BucketExists(bucketName)
+		if err != nil {
+			return nil, err
+		} else if !exists {
+			return nil, errors.New("could not create bucket and bucket does not exist, please check permissions: " + errMake.Error())
+		}
+	}
+
+	return &store{
+		client: client,
+		bucket: bucketName,
+	}, nil
+}
+
+func path(appName, callID string) string {
+	// raw url encode, b/c s3 does not like: & $ @ = : ; + , ?
+	appName = base64.RawURLEncoding.EncodeToString([]byte(appName)) // TODO optimize..
+	return appName + "/" + callID
+}
+
+func (s *store) InsertLog(ctx context.Context, appName, callID string, callLog io.Reader) error {
+	objectName := path(appName, callID)
+	_, err := s.client.PutObjectWithContext(ctx, s.bucket, objectName, callLog, -1, minio.PutObjectOptions{ContentType: contentType})
+	return err
+}
+
+func (s *store) GetLog(ctx context.Context, appName, callID string) (io.Reader, error) {
+	objectName := path(appName, callID)
+	obj, err := s.client.GetObjectWithContext(ctx, s.bucket, objectName, minio.GetObjectOptions{})
+	if err != nil {
+		return nil, err // this is always nil, for now, thanks minio :(
+	}
+
+	_, err = obj.Stat()
+	if err != nil {
+		errResp := minio.ToErrorResponse(err)
+		if errResp.StatusCode == 404 {
+			return nil, models.ErrCallLogNotFound
+		}
+		return nil, err
+	}
+
+	return obj, nil
+}

api/logs/s3/s3_test.go

@@ -0,0 +1,28 @@
+package s3
+
+import (
+	"net/url"
+	"os"
+	"testing"
+
+	logTesting "github.com/fnproject/fn/api/logs/testing"
+)
+
+func TestS3(t *testing.T) {
+	minio := os.Getenv("MINIO_URL")
+	if minio == "" {
+		t.Skip("no minio specified in url, skipping (use `make test`)")
+		return
+	}
+
+	uLog, err := url.Parse(minio)
+	if err != nil {
+		t.Fatalf("failed to parse url: %v", err)
+	}
+
+	ls, err := New(uLog)
+	if err != nil {
+		t.Fatalf("failed to create sqlite3 datastore: %v", err)
+	}
+	logTesting.Test(t, ls)
+}

api/logs/testing/test.go

@@ -1,7 +1,9 @@
 package testing
 
 import (
+	"bytes"
 	"context"
+	"io"
 	"strings"
 	"testing"
 	"time"
@@ -34,60 +36,32 @@ func SetupTestCall() *models.Call {
 	return &call
 }
 
-func Test(t *testing.T, fnl models.LogStore, ds models.Datastore) {
+func Test(t *testing.T, fnl models.LogStore) {
 	ctx := context.Background()
 	call := SetupTestCall()
 
-	t.Run("call-log-insert", func(t *testing.T) {
-		call.ID = id.New().String()
-		err := ds.InsertCall(ctx, call)
-		if err != nil {
-			t.Fatalf("Test InsertCall(ctx, &call): unexpected error `%v`", err)
-		}
-		log := strings.NewReader("test")
-		err = fnl.InsertLog(ctx, call.AppName, call.ID, log)
-		if err != nil {
-			t.Fatalf("Test InsertLog(ctx, call.ID, logText): unexpected error during inserting log `%v`", err)
-		}
-	})
 	t.Run("call-log-insert-get", func(t *testing.T) {
 		call.ID = id.New().String()
-		err := ds.InsertCall(ctx, call)
-		if err != nil {
-			t.Fatalf("Test InsertCall(ctx, &call): unexpected error `%v`", err)
-		}
 		logText := "test"
 		log := strings.NewReader(logText)
-		err = fnl.InsertLog(ctx, call.AppName, call.ID, log)
+		err := fnl.InsertLog(ctx, call.AppName, call.ID, log)
 		if err != nil {
 			t.Fatalf("Test InsertLog(ctx, call.ID, logText): unexpected error during inserting log `%v`", err)
 		}
 		logEntry, err := fnl.GetLog(ctx, call.AppName, call.ID)
-		if !strings.Contains(logEntry.Log, logText) {
+		var b bytes.Buffer
+		io.Copy(&b, logEntry)
+		if !strings.Contains(b.String(), logText) {
 			t.Fatalf("Test GetLog(ctx, call.ID, logText): unexpected error, log mismatch. "+
-				"Expected: `%v`. Got `%v`.", logText, logEntry.Log)
+				"Expected: `%v`. Got `%v`.", logText, b.String())
 		}
 	})
-	t.Run("call-log-insert-get-delete", func(t *testing.T) {
+
+	t.Run("call-log-not-found", func(t *testing.T) {
 		call.ID = id.New().String()
-		err := ds.InsertCall(ctx, call)
-		if err != nil {
-			t.Fatalf("Test InsertCall(ctx, &call): unexpected error `%v`", err)
-		}
-		logText := "test"
-		log := strings.NewReader(logText)
-		err = fnl.InsertLog(ctx, call.AppName, call.ID, log)
-		if err != nil {
-			t.Fatalf("Test InsertLog(ctx, call.ID, logText): unexpected error during inserting log `%v`", err)
-		}
-		logEntry, err := fnl.GetLog(ctx, call.AppName, call.ID)
-		if !strings.Contains(logEntry.Log, logText) {
-			t.Fatalf("Test GetLog(ctx, call.ID, logText): unexpected error, log mismatch. "+
-				"Expected: `%v`. Got `%v`.", logText, logEntry.Log)
-		}
-		err = fnl.DeleteLog(ctx, call.AppName, call.ID)
-		if err != nil {
-			t.Fatalf("Test DeleteLog(ctx, call.ID): unexpected error during deleting log `%v`", err)
+		_, err := fnl.GetLog(ctx, call.AppName, call.ID)
+		if err != models.ErrCallLogNotFound {
+			t.Fatal("GetLog should return not found, but got:", err)
 		}
 	})
 }

api/models/logs.go

@@ -12,10 +12,11 @@ type LogStore interface {
 
 	// GetLog will return the log at callID, an error will be returned if the log
 	// cannot be found.
-	// TODO it would be nice if this were an io.Reader...
-	GetLog(ctx context.Context, appName, callID string) (*CallLog, error)
+	GetLog(ctx context.Context, appName, callID string) (io.Reader, error)
 
-	// DeleteLog will remove the log at callID, it will not return an error if
-	// the log does not exist before removal.
-	DeleteLog(ctx context.Context, appName, callID string) error
+	// TODO we should probably allow deletion of a range of logs (also calls)?
+	// common cases for deletion will be:
+	// * route gets nuked
+	// * app gets nuked
+	// * call+logs getting cleaned up periodically
 }

api/server/call_logs.go

@@ -1,9 +1,11 @@
 package server
 
 import (
+	"bytes"
 	"net/http"
 
 	"github.com/fnproject/fn/api"
+	"github.com/fnproject/fn/api/models"
 	"github.com/gin-gonic/gin"
 )
 
@@ -12,35 +14,25 @@ func (s *Server) handleCallLogGet(c *gin.Context) {
 
 	appName := c.MustGet(api.AppName).(string)
 	callID := c.Param(api.Call)
-	_, err := s.Datastore.GetCall(ctx, appName, callID)
+
+	logReader, err := s.LogDB.GetLog(ctx, appName, callID)
 	if err != nil {
 		handleErrorResponse(c, err)
 		return
 	}
 
-	callObj, err := s.LogDB.GetLog(ctx, appName, callID)
-	if err != nil {
-		handleErrorResponse(c, err)
-		return
+	// TODO this API needs to change to text/plain / gzip anyway, punting
+	// optimization, but we can write this direct to the wire, too... seems like
+	// we should write some kind of writev json thing for go since we keep
+	// hitting this :(
+	var b bytes.Buffer
+	b.ReadFrom(logReader)
+
+	callObj := models.CallLog{
+		CallID:  callID,
+		AppName: appName,
+		Log:     b.String(),
 	}
 
-	c.JSON(http.StatusOK, callLogResponse{"Successfully loaded call", callObj})
-}
-
-func (s *Server) handleCallLogDelete(c *gin.Context) {
-	ctx := c.Request.Context()
-
-	appName := c.MustGet(api.AppName).(string)
-	callID := c.Param(api.Call)
-	_, err := s.Datastore.GetCall(ctx, appName, callID)
-	if err != nil {
-		handleErrorResponse(c, err)
-		return
-	}
-	err = s.LogDB.DeleteLog(ctx, appName, callID)
-	if err != nil {
-		handleErrorResponse(c, err)
-		return
-	}
-	c.JSON(http.StatusAccepted, gin.H{"message": "Log delete accepted"})
+	c.JSON(http.StatusOK, callLogResponse{"Successfully loaded log", &callObj})
 }

api/server/runner_test.go

@@ -27,7 +27,7 @@ func testRunner(t *testing.T, args ...interface{}) (agent.Agent, context.CancelF
 			mq = arg
 		}
 	}
-	r := agent.New(ds, mq)
+	r := agent.New(ds, ds, mq)
 	return r, func() { r.Close() }
 }
 

api/server/server.go

@@ -95,16 +95,15 @@ func optionalCorsWrap(r *gin.Engine) {
 }
 
 // New creates a new Functions server with the passed in datastore, message queue and API URL
-func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, logDB models.LogStore, opts ...ServerOption) *Server {
-
+func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, ls models.LogStore, opts ...ServerOption) *Server {
 	setTracer()
 
 	s := &Server{
-		Agent:     agent.New(cache.Wrap(ds), mq), // only add datastore caching to agent
+		Agent:     agent.New(cache.Wrap(ds), ls, mq), // only add datastore caching to agent
 		Router:    gin.New(),
 		Datastore: ds,
 		MQ:        mq,
-		LogDB:     logDB,
+		LogDB:     ls,
 	}
 
 	setMachineID()
@@ -355,7 +354,6 @@ func (s *Server) bindHandlers(ctx context.Context) {
 
 			apps.GET("/calls/:call", s.handleCallGet)
 			apps.GET("/calls/:call/log", s.handleCallLogGet)
-			apps.DELETE("/calls/:call/log", s.handleCallLogDelete)
 		}
 	}