diff --git a/canary/looper.go b/canary/looper.go index ce7fef99..6c59b766 100644 --- a/canary/looper.go +++ b/canary/looper.go @@ -1,20 +1,30 @@ package canary import ( + "bytes" + "github.com/openziti/sdk-golang/ziti" + "github.com/openziti/sdk-golang/ziti/edge" "github.com/openziti/zrok/environment/env_core" "github.com/openziti/zrok/sdk/golang/sdk" + "github.com/pkg/errors" "github.com/sirupsen/logrus" + "io" + "math/rand" + "net/http" + "time" ) type PublicHttpLooper struct { id int frontend string + opt *LooperOptions root env_core.Root shr *sdk.Share + listener edge.Listener done chan struct{} } -func NewPublicHttpLooper(id int, frontend string, root env_core.Root) *PublicHttpLooper { +func NewPublicHttpLooper(id int, frontend string, opt *LooperOptions, root env_core.Root) *PublicHttpLooper { return &PublicHttpLooper{ id: id, frontend: frontend, @@ -29,10 +39,16 @@ func (l *PublicHttpLooper) Run() { logrus.Infof("starting #%d", l.id) if err := l.startup(); err != nil { - logrus.Fatalf("error starting #%d: %v", l.id, err) + logrus.Fatalf("#%d error starting: %v", l.id, err) } - logrus.Infof("#%d complete") + if err := l.bindListener(); err != nil { + logrus.Fatalf("#%d error binding listener: %v", l.id, err) + } + + l.dwell() + + logrus.Infof("completed #%d", l.id) if err := l.shutdown(); err != nil { logrus.Fatalf("error shutting down #%d: %v", l.id, err) } @@ -55,6 +71,62 @@ func (l *PublicHttpLooper) startup() error { return nil } +func (l *PublicHttpLooper) bindListener() error { + zif, err := l.root.ZitiIdentityNamed(l.root.EnvironmentIdentityName()) + if err != nil { + return errors.Wrapf(err, "#%d error getting identity", l.id) + } + zcfg, err := ziti.NewConfigFromFile(zif) + if err != nil { + return errors.Wrapf(err, "#%d error loading ziti config", l.id) + } + options := ziti.ListenOptions{ + ConnectTimeout: 5 * time.Minute, + WaitForNEstablishedListeners: 1, + } + zctx, err := ziti.NewContext(zcfg) + if err != nil { + return errors.Wrapf(err, "#%d error creating ziti context", l.id) + } + + if l.listener, err = zctx.ListenWithOptions(l.shr.Token, &options); err != nil { + return errors.Wrapf(err, "#%d error binding listener", l.id) + } + + go func() { + if err := http.Serve(l.listener, l); err != nil { + logrus.Errorf("#%d error starting http listener: %v", l.id, err) + } + }() + + return nil +} + +func (l *PublicHttpLooper) ServeHTTP(w http.ResponseWriter, r *http.Request) { + buf := new(bytes.Buffer) + io.Copy(buf, r.Body) + w.Write(buf.Bytes()) +} + +func (l *PublicHttpLooper) dwell() { + dwell := l.opt.MinDwell.Milliseconds() + dwelta := l.opt.MaxDwell.Milliseconds() - l.opt.MinDwell.Milliseconds() + if dwelta > 0 { + dwell = int64(rand.Intn(int(dwelta)) + int(l.opt.MinDwell.Milliseconds())) + } + time.Sleep(time.Duration(dwell) * time.Millisecond) +} + func (l *PublicHttpLooper) shutdown() error { + if l.listener != nil { + if err := l.listener.Close(); err != nil { + logrus.Errorf("#%d error closing listener: %v", l.id, err) + } + } + + if err := sdk.DeleteShare(l.root, l.shr); err != nil { + return errors.Wrapf(err, "#%d error deleting share", l.id) + } + return nil } diff --git a/canary/looperOptions.go b/canary/looperOptions.go new file mode 100644 index 00000000..7d47aecc --- /dev/null +++ b/canary/looperOptions.go @@ -0,0 +1,14 @@ +package canary + +import "time" + +type LooperOptions struct { + Iterations uint + Timeout time.Duration + MinPayload uint64 + MaxPayload uint64 + MinDwell time.Duration + MaxDwell time.Duration + MinPacing time.Duration + MaxPacing time.Duration +}