alihan/fn-serverless
1
0
Fork 0
mirror of https://github.com/fnproject/fn.git synced 2022-10-28 21:29:17 +03:00
Code Issues Packages Projects Releases Wiki Activity

Hot fn ID

Browse Source
This commit is contained in:
Denis Makogon
2017-05-31 22:50:31 -07:00
parent 102d025332
commit 007f1982fc
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
api/runner/worker.go
View File

@@ -9,6 +9,7 @@ import (
"time" "time"
"github.com/Sirupsen/logrus" "github.com/Sirupsen/logrus"
uuid "github.com/satori/go.uuid"
"gitlab-odx.oracle.com/odx/functions/api/runner/drivers" "gitlab-odx.oracle.com/odx/functions/api/runner/drivers"
"gitlab-odx.oracle.com/odx/functions/api/runner/protocol" "gitlab-odx.oracle.com/odx/functions/api/runner/protocol"
"gitlab-odx.oracle.com/odx/functions/api/runner/task" "gitlab-odx.oracle.com/odx/functions/api/runner/task"
@@ -214,6 +215,7 @@ func (svr *htfnsvr) launch(ctx context.Context) error {
// stream into a long lived container. If idle long enough, it will stop. It // stream into a long lived container. If idle long enough, it will stop. It
// uses route configuration to determine which protocol to use. // uses route configuration to determine which protocol to use.
type htfn struct { type htfn struct {
id string
cfg *task.Config cfg *task.Config
proto protocol.ContainerIO proto protocol.ContainerIO
tasks <-chan task.Request tasks <-chan task.Request
@@ -240,6 +242,7 @@ func newhtfn(cfg *task.Config, proto protocol.Protocol, tasks <-chan task.Reques
} }
hc := &htfn{ hc := &htfn{
id: uuid.NewV5(uuid.Nil, fmt.Sprintf("%s%s%d", cfg.AppName, cfg.Path, time.Now().Unix())).String(),
cfg: cfg, cfg: cfg,
proto: p, proto: p,
tasks: tasks, tasks: tasks,
@@ -260,7 +263,8 @@ func (hc *htfn) serve(ctx context.Context) {
lctx, cancel := context.WithCancel(ctx) lctx, cancel := context.WithCancel(ctx)
var wg sync.WaitGroup var wg sync.WaitGroup
cfg := *hc.cfg cfg := *hc.cfg
logger := logrus.WithFields(logrus.Fields{ logFields := logrus.Fields{
"hot_id": hc.id,
"app": cfg.AppName, "app": cfg.AppName,
"route": cfg.Path, "route": cfg.Path,
"image": cfg.Image, "image": cfg.Image,
@@ -268,7 +272,8 @@ func (hc *htfn) serve(ctx context.Context) {
"format": cfg.Format, "format": cfg.Format,
"max_concurrency": cfg.MaxConcurrency, "max_concurrency": cfg.MaxConcurrency,
"idle_timeout": cfg.IdleTimeout, "idle_timeout": cfg.IdleTimeout,
}) }
logger := logrus.WithFields(logFields)
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -336,11 +341,11 @@ func (hc *htfn) serve(ctx context.Context) {
result, err := hc.rnr.Run(lctx, &cfg) result, err := hc.rnr.Run(lctx, &cfg)
if err != nil { if err != nil {
logrus.WithError(err).Error("hot function failure detected") logger.WithError(err).Error("hot function failure detected")
} }
errw.Close() errw.Close()
wg.Wait() wg.Wait()
logrus.WithField("result", result).Info("hot function terminated") logger.WithField("result", result).Info("hot function terminated")
} }
func runTaskReq(rnr *Runner, wg *sync.WaitGroup, t task.Request) { func runTaskReq(rnr *Runner, wg *sync.WaitGroup, t task.Request) {