Merge pull request #290 from fnproject/issue-276

Use context-bound SQL methods

api/agent/agent.go

@@ -222,9 +222,12 @@ func (a *agent) Submit(callI Call) error {
 
 	a.stats.Complete()
 
-	// TODO if the context is timed out here we need to allocate some more time...
-	// right now this only works b/c the db isn't using the context
-	return call.End(ctx, err)
+	// TODO: we need to allocate more time to store the call + logs in case the call timed out,
+	// but this could put us over the timeout if the call did not reply yet (need better policy).
+	ctx = opentracing.ContextWithSpan(context.Background(), span)
+	call.End(ctx, err)
+
+	return err
 }
 
 // getSlot must ensure that if it receives a slot, it will be returned, otherwise

api/agent/call.go

@@ -35,7 +35,7 @@ type Call interface {
 	// regardless of whether the execution failed or not. An error will be passed
 	// to End, which if nil indicates a successful execution. Any error returned
 	// from End will be returned as the error from Submit.
-	End(ctx context.Context, err error) error
+	End(ctx context.Context, err error)
 }
 
 // TODO build w/o closures... lazy
@@ -267,7 +267,7 @@ func (c *call) Start(ctx context.Context) error {
 	return nil
 }
 
-func (c *call) End(ctx context.Context, err error) error {
+func (c *call) End(ctx context.Context, err error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "agent_call_end")
 	defer span.Finish()
 
@@ -287,19 +287,12 @@ func (c *call) End(ctx context.Context, err error) error {
 		// XXX (reed): delete MQ message, eventually
 	}
 
-	// TODO since the function itself can reply directly to a client (or logs),
 	// this means that we could potentially store an error / timeout status for a
 	// call that ran successfully [by a user's perspective]
-	// TODO this should be update, really
+	// TODO: this should be update, really
 	if err := c.ds.InsertCall(ctx, c.Call); err != nil {
-		// TODO we should just log this error not return it to user? just issue storing call status but call is run
 		logrus.WithError(err).Error("error inserting call into datastore")
 	}
-
-	// return the original error so that this is returned from Submit (for sync)
-	// TODO we could just skip over (and log) and End errors and return slot.exec error
-	// in submit instead of doing this, it's a bit odd. thoughts?
-	return err
 }
 
 func (a *agent) route(ctx context.Context, appName, path string) (*models.Route, error) {

api/agent/func_logger.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/fnproject/fn/api/common"
 	"github.com/fnproject/fn/api/models"
+	"github.com/opentracing/opentracing-go"
 	"github.com/sirupsen/logrus"
 )
 
@@ -56,7 +57,7 @@ func NewFuncLogger(ctx context.Context, appName, path, image, reqID string, logD
 		reqID:  reqID,
 	})
 
-	// TODO / NOTE: we want linew to be first becauase limitw may error if limit
+	// TODO / NOTE: we want linew to be first because limitw may error if limit
 	// is reached but we still want to log. we should probably ignore hitting the
 	// limit error since we really just want to not write too much to db and
 	// that's handled as is. put buffers back last to avoid misuse, if there's
@@ -110,7 +111,7 @@ type logWriter struct {
 func (l *logWriter) Write(b []byte) (int, error) {
 	log := common.Logger(l.ctx)
 	log = log.WithFields(logrus.Fields{"user_log": true, "app_name": l.appName, "path": l.path, "image": l.image, "call_id": l.reqID})
-	log.Println(string(b))
+	log.Debug(string(b))
 	return len(b), nil
 }
 
@@ -182,7 +183,9 @@ type dbWriter struct {
 }
 
 func (w *dbWriter) Close() error {
-	return w.db.InsertLog(w.ctx, w.reqID, w.String())
+	span, ctx := opentracing.StartSpanFromContext(context.Background(), "agent_log_write")
+	defer span.Finish()
+	return w.db.InsertLog(ctx, w.reqID, w.String())
 }
 
 func (w *dbWriter) Write(b []byte) (int, error) {

api/datastore/internal/datastoretest/test.go

@@ -13,9 +13,9 @@ import (
 	"reflect"
 	"time"
 
-	"github.com/sirupsen/logrus"
 	"github.com/gin-gonic/gin"
 	"github.com/go-openapi/strfmt"
+	"github.com/sirupsen/logrus"
 )
 
 func setLogBuffer() *bytes.Buffer {

api/datastore/sql/sql.go

@@ -12,7 +12,6 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/sirupsen/logrus"
 	"github.com/fnproject/fn/api/models"
 	"github.com/go-sql-driver/mysql"
 	_ "github.com/go-sql-driver/mysql"
@@ -21,6 +20,7 @@ import (
 	_ "github.com/lib/pq"
 	"github.com/mattn/go-sqlite3"
 	_ "github.com/mattn/go-sqlite3"
+	"github.com/sirupsen/logrus"
 )
 
 // this aims to be an ANSI-SQL compliant package that uses only question
@@ -147,7 +147,7 @@ func (ds *sqlStore) InsertApp(ctx context.Context, app *models.App) (*models.App
 	}
 
 	query := ds.db.Rebind("INSERT INTO apps (name, config) VALUES (?, ?);")
-	_, err = ds.db.Exec(query, app.Name, string(cbyte))
+	_, err = ds.db.ExecContext(ctx, query, app.Name, string(cbyte))
 	if err != nil {
 		switch err := err.(type) {
 		case *mysql.MySQLError:
@@ -173,7 +173,7 @@ func (ds *sqlStore) UpdateApp(ctx context.Context, newapp *models.App) (*models.
 	app := &models.App{Name: newapp.Name}
 	err := ds.Tx(func(tx *sqlx.Tx) error {
 		query := tx.Rebind(`SELECT config FROM apps WHERE name=?`)
-		row := tx.QueryRow(query, app.Name)
+		row := tx.QueryRowContext(ctx, query, app.Name)
 
 		var config string
 		if err := row.Scan(&config); err != nil {
@@ -198,7 +198,7 @@ func (ds *sqlStore) UpdateApp(ctx context.Context, newapp *models.App) (*models.
 		}
 
 		query = tx.Rebind(`UPDATE apps SET config=? WHERE name=?`)
-		res, err := tx.Exec(query, string(cbyte), app.Name)
+		res, err := tx.ExecContext(ctx, query, string(cbyte), app.Name)
 		if err != nil {
 			return err
 		}
@@ -221,13 +221,13 @@ func (ds *sqlStore) UpdateApp(ctx context.Context, newapp *models.App) (*models.
 
 func (ds *sqlStore) RemoveApp(ctx context.Context, appName string) error {
 	query := ds.db.Rebind(`DELETE FROM apps WHERE name = ?`)
-	_, err := ds.db.Exec(query, appName)
+	_, err := ds.db.ExecContext(ctx, query, appName)
 	return err
 }
 
 func (ds *sqlStore) GetApp(ctx context.Context, name string) (*models.App, error) {
 	query := ds.db.Rebind(`SELECT name, config FROM apps WHERE name=?`)
-	row := ds.db.QueryRow(query, name)
+	row := ds.db.QueryRowContext(ctx, query, name)
 
 	var resName, config string
 	err := row.Scan(&resName, &config)
@@ -257,7 +257,7 @@ func (ds *sqlStore) GetApps(ctx context.Context, filter *models.AppFilter) ([]*m
 	res := []*models.App{}
 	query, args := buildFilterAppQuery(filter)
 	query = ds.db.Rebind(fmt.Sprintf("SELECT DISTINCT name, config FROM apps %s", query))
-	rows, err := ds.db.Query(query, args...)
+	rows, err := ds.db.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -295,14 +295,14 @@ func (ds *sqlStore) InsertRoute(ctx context.Context, route *models.Route) (*mode
 
 	err = ds.Tx(func(tx *sqlx.Tx) error {
 		query := tx.Rebind(`SELECT 1 FROM apps WHERE name=?`)
-		r := tx.QueryRow(query, route.AppName)
+		r := tx.QueryRowContext(ctx, query, route.AppName)
 		if err := r.Scan(new(int)); err != nil {
 			if err == sql.ErrNoRows {
 				return models.ErrAppsNotFound
 			}
 		}
 		query = tx.Rebind(`SELECT 1 FROM routes WHERE app_name=? AND path=?`)
-		same, err := tx.Query(query, route.AppName, route.Path)
+		same, err := tx.QueryContext(ctx, query, route.AppName, route.Path)
 		if err != nil {
 			return err
 		}
@@ -325,7 +325,7 @@ func (ds *sqlStore) InsertRoute(ctx context.Context, route *models.Route) (*mode
 		)
 		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);`)
 
-		_, err = tx.Exec(query,
+		_, err = tx.ExecContext(ctx, query,
 			route.AppName,
 			route.Path,
 			route.Image,
@@ -348,7 +348,7 @@ func (ds *sqlStore) UpdateRoute(ctx context.Context, newroute *models.Route) (*m
 	var route models.Route
 	err := ds.Tx(func(tx *sqlx.Tx) error {
 		query := tx.Rebind(fmt.Sprintf("%s WHERE app_name=? AND path=?", routeSelector))
-		row := tx.QueryRow(query, newroute.AppName, newroute.Path)
+		row := tx.QueryRowContext(ctx, query, newroute.AppName, newroute.Path)
 		if err := scanRoute(row, &route); err == sql.ErrNoRows {
 			return models.ErrRoutesNotFound
 		} else if err != nil {
@@ -378,7 +378,7 @@ func (ds *sqlStore) UpdateRoute(ctx context.Context, newroute *models.Route) (*m
 			config = ?
 		WHERE app_name=? AND path=?;`)
 
-		res, err := tx.Exec(query,
+		res, err := tx.ExecContext(ctx, query,
 			route.Image,
 			route.Format,
 			route.Memory,
@@ -413,7 +413,7 @@ func (ds *sqlStore) UpdateRoute(ctx context.Context, newroute *models.Route) (*m
 
 func (ds *sqlStore) RemoveRoute(ctx context.Context, appName, routePath string) error {
 	query := ds.db.Rebind(`DELETE FROM routes WHERE path = ? AND app_name = ?`)
-	res, err := ds.db.Exec(query, routePath, appName)
+	res, err := ds.db.ExecContext(ctx, query, routePath, appName)
 	if err != nil {
 		return err
 	}
@@ -433,7 +433,7 @@ func (ds *sqlStore) RemoveRoute(ctx context.Context, appName, routePath string)
 func (ds *sqlStore) GetRoute(ctx context.Context, appName, routePath string) (*models.Route, error) {
 	rSelectCondition := "%s WHERE app_name=? AND path=?"
 	query := ds.db.Rebind(fmt.Sprintf(rSelectCondition, routeSelector))
-	row := ds.db.QueryRow(query, appName, routePath)
+	row := ds.db.QueryRowContext(ctx, query, appName, routePath)
 
 	var route models.Route
 	err := scanRoute(row, &route)
@@ -451,7 +451,7 @@ func (ds *sqlStore) GetRoutes(ctx context.Context, filter *models.RouteFilter) (
 	query, args := buildFilterRouteQuery(filter)
 	query = fmt.Sprintf("%s %s", routeSelector, query)
 	query = ds.db.Rebind(query)
-	rows, err := ds.db.Query(query, args...)
+	rows, err := ds.db.QueryContext(ctx, query, args...)
 	// todo: check for no rows so we don't respond with a sql 500 err
 	if err != nil {
 		return nil, err
@@ -490,7 +490,7 @@ func (ds *sqlStore) GetRoutesByApp(ctx context.Context, appName string, filter *
 
 	query := fmt.Sprintf("%s %s", routeSelector, filterQuery)
 	query = ds.db.Rebind(query)
-	rows, err := ds.db.Query(query, args...)
+	rows, err := ds.db.QueryContext(ctx, query, args...)
 	// todo: check for no rows so we don't respond with a sql 500 err
 	if err != nil {
 		return nil, err
@@ -538,7 +538,7 @@ func (ds *sqlStore) InsertCall(ctx context.Context, call *models.Call) error {
 	)
 	VALUES (?, ?, ?, ?, ?, ?, ?);`)
 
-	_, err := ds.db.Exec(query, call.ID, call.CreatedAt.String(),
+	_, err := ds.db.ExecContext(ctx, query, call.ID, call.CreatedAt.String(),
 		call.StartedAt.String(), call.CompletedAt.String(),
 		call.Status, call.AppName, call.Path)
 	if err != nil {
@@ -554,7 +554,7 @@ func (ds *sqlStore) InsertCall(ctx context.Context, call *models.Call) error {
 func (ds *sqlStore) GetCall(ctx context.Context, appName, callID string) (*models.Call, error) {
 	query := fmt.Sprintf(`%s WHERE id=? AND app_name=?`, callSelector)
 	query = ds.db.Rebind(query)
-	row := ds.db.QueryRow(query, callID, appName)
+	row := ds.db.QueryRowContext(ctx, query, callID, appName)
 
 	var call models.Call
 	err := scanCall(row, &call)
@@ -569,7 +569,7 @@ func (ds *sqlStore) GetCalls(ctx context.Context, filter *models.CallFilter) ([]
 	query, args := buildFilterCallQuery(filter)
 	query = fmt.Sprintf("%s %s", callSelector, query)
 	query = ds.db.Rebind(query)
-	rows, err := ds.db.Query(query, args...)
+	rows, err := ds.db.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -591,13 +591,13 @@ func (ds *sqlStore) GetCalls(ctx context.Context, filter *models.CallFilter) ([]
 
 func (ds *sqlStore) InsertLog(ctx context.Context, callID, callLog string) error {
 	query := ds.db.Rebind(`INSERT INTO logs (id, log) VALUES (?, ?);`)
-	_, err := ds.db.Exec(query, callID, callLog)
+	_, err := ds.db.ExecContext(ctx, query, callID, callLog)
 	return err
 }
 
 func (ds *sqlStore) GetLog(ctx context.Context, callID string) (*models.CallLog, error) {
 	query := ds.db.Rebind(`SELECT log FROM logs WHERE id=?`)
-	row := ds.db.QueryRow(query, callID)
+	row := ds.db.QueryRowContext(ctx, query, callID)
 
 	var log string
 	err := row.Scan(&log)
@@ -616,7 +616,7 @@ func (ds *sqlStore) GetLog(ctx context.Context, callID string) (*models.CallLog,
 
 func (ds *sqlStore) DeleteLog(ctx context.Context, callID string) error {
 	query := ds.db.Rebind(`DELETE FROM logs WHERE id=?`)
-	_, err := ds.db.Exec(query, callID)
+	_, err := ds.db.ExecContext(ctx, query, callID)
 	return err
 }
 

api/logs/log.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"net/url"
 
-	"github.com/sirupsen/logrus"
 	"github.com/fnproject/fn/api/datastore/sql"
 	"github.com/fnproject/fn/api/models"
+	"github.com/sirupsen/logrus"
 )
 
 func New(dbURL string) (models.LogStore, error) {

api/mqs/bolt.go

@@ -11,10 +11,10 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/sirupsen/logrus"
 	"github.com/boltdb/bolt"
 	"github.com/fnproject/fn/api/common"
 	"github.com/fnproject/fn/api/models"
+	"github.com/sirupsen/logrus"
 )
 
 type BoltDbMQ struct {

api/mqs/ironmq.go

@@ -9,10 +9,10 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/sirupsen/logrus"
 	"github.com/fnproject/fn/api/models"
 	mq_config "github.com/iron-io/iron_go3/config"
 	ironmq "github.com/iron-io/iron_go3/mq"
+	"github.com/sirupsen/logrus"
 )
 
 type assoc struct {

api/mqs/new.go

@@ -6,9 +6,9 @@ import (
 	"net/url"
 	"strings"
 
-	"github.com/sirupsen/logrus"
 	"github.com/fnproject/fn/api/models"
 	"github.com/opentracing/opentracing-go"
+	"github.com/sirupsen/logrus"
 )
 
 // New will parse the URL and return the correct MQ implementation.

api/server/apps_test.go

@@ -7,12 +7,12 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/sirupsen/logrus"
 	"github.com/fnproject/fn/api/datastore"
 	"github.com/fnproject/fn/api/logs"
 	"github.com/fnproject/fn/api/models"
 	"github.com/fnproject/fn/api/mqs"
 	"github.com/gin-gonic/gin"
+	"github.com/sirupsen/logrus"
 )
 
 func setLogBuffer() *bytes.Buffer {

api/server/init.go

@@ -7,8 +7,8 @@ import (
 	"os/signal"
 	"strings"
 
-	"github.com/sirupsen/logrus"
 	"github.com/gin-gonic/gin"
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )
 

fnlb/lb/allgrouper.go

@@ -17,12 +17,12 @@ import (
 	"time"
 
 	"fmt"
-	"github.com/sirupsen/logrus"
 	"github.com/coreos/go-semver/semver"
 	"github.com/go-sql-driver/mysql"
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/mattn/go-sqlite3"
+	"github.com/sirupsen/logrus"
 )
 
 // NewAllGrouper returns a Grouper that will return the entire list of nodes

fnlb/lb/proxy.go

@@ -7,11 +7,11 @@ import (
 	"net/http/httputil"
 	"sync"
 
-	"github.com/sirupsen/logrus"
 	"github.com/coreos/go-semver/semver"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	"github.com/openzipkin/zipkin-go-opentracing"
+	"github.com/sirupsen/logrus"
 )
 
 // TODO the load balancers all need to have the same list of nodes. gossip?

fnlb/main.go

@@ -12,9 +12,9 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/sirupsen/logrus"
 	"github.com/coreos/go-semver/semver"
 	"github.com/fnproject/fn/fnlb/lb"
+	"github.com/sirupsen/logrus"
 )
 
 const VERSION = "0.0.56"