mask secrets, backport drone_ variables

This commit is contained in:
Brad Rydzewski 2017-03-12 00:46:59 -08:00
parent b4c4e92b5b
commit 442e05a4e1
13 changed files with 297 additions and 115 deletions

View file

@ -26,6 +26,11 @@ func loop(c *cli.Context) error {
if err != nil { if err != nil {
return err return err
} }
filter := rpc.Filter{
Labels: map[string]string{
"platform": c.String("platform"),
},
}
client, err := rpc.NewClient( client, err := rpc.NewClient(
endpoint.String(), endpoint.String(),
@ -62,7 +67,7 @@ func loop(c *cli.Context) error {
if sigterm.IsSet() { if sigterm.IsSet() {
return return
} }
if err := run(ctx, client); err != nil { if err := run(ctx, client, filter); err != nil {
log.Printf("build runner encountered error: exiting: %s", err) log.Printf("build runner encountered error: exiting: %s", err)
return return
} }
@ -74,11 +79,16 @@ func loop(c *cli.Context) error {
return nil return nil
} }
func run(ctx context.Context, client rpc.Peer) error { const (
maxFileUpload = 5000000
maxLogsUpload = 5000000
)
func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
log.Println("pipeline: request next execution") log.Println("pipeline: request next execution")
// get the next job from the queue // get the next job from the queue
work, err := client.Next(ctx) work, err := client.Next(ctx, filter)
if err != nil { if err != nil {
return err return err
} }
@ -103,9 +113,9 @@ func run(ctx context.Context, client rpc.Peer) error {
cancelled := abool.New() cancelled := abool.New()
go func() { go func() {
if err := client.Wait(ctx, work.ID); err != nil { if werr := client.Wait(ctx, work.ID); err != nil {
cancelled.SetTo(true) cancelled.SetTo(true)
log.Printf("pipeline: cancel signal received: %s: %s", work.ID, err) log.Printf("pipeline: cancel signal received: %s: %s", work.ID, werr)
cancel() cancel()
} else { } else {
log.Printf("pipeline: cancel channel closed: %s", work.ID) log.Printf("pipeline: cancel channel closed: %s", work.ID)
@ -140,7 +150,8 @@ func run(ctx context.Context, client rpc.Peer) error {
} }
uploads.Add(1) uploads.Add(1)
writer := rpc.NewLineWriter(client, work.ID, proc.Alias) writer := rpc.NewLineWriter(client, work.ID, proc.Alias)
io.Copy(writer, part) rlimit := io.LimitReader(part, maxLogsUpload)
io.Copy(writer, rlimit)
defer func() { defer func() {
log.Printf("pipeline: finish uploading logs: %s: step %s", work.ID, proc.Alias) log.Printf("pipeline: finish uploading logs: %s: step %s", work.ID, proc.Alias)
@ -151,8 +162,9 @@ func run(ctx context.Context, client rpc.Peer) error {
if rerr != nil { if rerr != nil {
return nil return nil
} }
rlimit = io.LimitReader(part, maxFileUpload)
mime := part.Header().Get("Content-Type") mime := part.Header().Get("Content-Type")
if serr := client.Save(context.Background(), work.ID, mime, part); serr != nil { if serr := client.Upload(context.Background(), work.ID, mime, rlimit); serr != nil {
log.Printf("pipeline: cannot upload artifact: %s: %s: %s", work.ID, mime, serr) log.Printf("pipeline: cannot upload artifact: %s: %s: %s", work.ID, mime, serr)
} }
return nil return nil

View file

@ -249,7 +249,7 @@ func PostHook2(c *gin.Context) {
for _, job := range jobs { for _, job := range jobs {
metadata := metadataFromStruct(repo, build, last, job, "linux/amd64") metadata := metadataFromStruct(repo, build, last, job, httputil.GetURL(c.Request))
environ := metadata.Environ() environ := metadata.Environ()
secrets := map[string]string{} secrets := map[string]string{}
@ -296,6 +296,7 @@ func PostHook2(c *gin.Context) {
ir := compiler.New( ir := compiler.New(
compiler.WithEnviron(environ), compiler.WithEnviron(environ),
// TODO ability to customize the escalated plugins
compiler.WithEscalated("plugins/docker", "plugins/gcr", "plugins/ecr"), compiler.WithEscalated("plugins/docker", "plugins/gcr", "plugins/ecr"),
compiler.WithLocal(false), compiler.WithLocal(false),
compiler.WithNetrc(netrc.Login, netrc.Password, netrc.Machine), compiler.WithNetrc(netrc.Login, netrc.Password, netrc.Machine),
@ -306,17 +307,46 @@ func PostHook2(c *gin.Context) {
time.Now().Unix(), time.Now().Unix(),
), ),
), ),
compiler.WithEnviron(job.Environment),
compiler.WithProxy(), compiler.WithProxy(),
compiler.WithVolumes(), // todo set global volumes // TODO ability to set global volumes for things like certs
compiler.WithVolumes(),
compiler.WithWorkspaceFromURL("/drone", repo.Link), compiler.WithWorkspaceFromURL("/drone", repo.Link),
).Compile(parsed) ).Compile(parsed)
// TODO there is a chicken and egg problem here because
// the compiled yaml has a platform environment variable
// that is not correctly set, because we are just about
// to set it ....
// TODO maybe we remove platform from metadata and let
// the compiler set the value from the yaml itself.
if parsed.Platform == "" {
parsed.Platform = "linux/amd64"
}
for _, sec := range secs {
if !sec.MatchEvent(build.Event) {
continue
}
if build.Verified || sec.SkipVerify {
ir.Secrets = append(ir.Secrets, &backend.Secret{
Mask: sec.Conceal,
Name: sec.Name,
Value: sec.Value,
})
}
}
task := new(queue.Task) task := new(queue.Task)
task.ID = fmt.Sprint(job.ID) task.ID = fmt.Sprint(job.ID)
task.Labels = map[string]string{} task.Labels = map[string]string{}
task.Labels["platform"] = "linux/amd64" task.Labels["platform"] = parsed.Platform
// TODO set proper platform if parsed.Labels != nil {
// TODO set proper labels for k, v := range parsed.Labels {
task.Labels[k] = v
}
}
task.Data, _ = json.Marshal(rpc.Pipeline{ task.Data, _ = json.Marshal(rpc.Pipeline{
ID: fmt.Sprint(job.ID), ID: fmt.Sprint(job.ID),
Config: ir, Config: ir,

View file

@ -78,9 +78,16 @@ type RPC struct {
} }
// Next implements the rpc.Next function // Next implements the rpc.Next function
func (s *RPC) Next(c context.Context) (*rpc.Pipeline, error) { func (s *RPC) Next(c context.Context, filter rpc.Filter) (*rpc.Pipeline, error) {
filter := func(*queue.Task) bool { return true } fn := func(task *queue.Task) bool {
task, err := s.queue.Poll(c, filter) for k, v := range filter.Labels {
if task.Labels[k] != v {
return false
}
}
return true
}
task, err := s.queue.Poll(c, fn)
if err != nil { if err != nil {
return nil, err return nil, err
} else if task == nil { } else if task == nil {
@ -207,8 +214,8 @@ func (s *RPC) Update(c context.Context, id string, state rpc.State) error {
return nil return nil
} }
// Save implements the rpc.Save function // Upload implements the rpc.Upload function
func (s *RPC) Save(c context.Context, id, mime string, file io.Reader) error { return nil } func (s *RPC) Upload(c context.Context, id, mime string, file io.Reader) error { return nil }
// Done implements the rpc.Done function // Done implements the rpc.Done function
func (s *RPC) Done(c context.Context, id string) error { return nil } func (s *RPC) Done(c context.Context, id string) error { return nil }

View file

@ -6,6 +6,7 @@ type (
Stages []*Stage `json:"pipeline"` // pipeline stages Stages []*Stage `json:"pipeline"` // pipeline stages
Networks []*Network `json:"networks"` // network definitions Networks []*Network `json:"networks"` // network definitions
Volumes []*Volume `json:"volumes"` // volume definitions Volumes []*Volume `json:"volumes"` // volume definitions
Secrets []*Secret `json:"secrets"` // secret definitions
} }
// Stage denotes a collection of one or more steps. // Stage denotes a collection of one or more steps.
@ -72,6 +73,14 @@ type (
DriverOpts map[string]string `json:"driver_opts,omitempty"` DriverOpts map[string]string `json:"driver_opts,omitempty"`
} }
// Secret defines a runtime secret
Secret struct {
Name string `json:"name,omitempty"`
Value string `json:"value,omitempty"`
Mount string `json:"mount,omitempty"`
Mask bool `json:"mask,omitempty"`
}
// State defines a container state. // State defines a container state.
State struct { State struct {
// Container exit code // Container exit code

View file

@ -1,6 +1,19 @@
package frontend package frontend
import "strconv" import (
"fmt"
"regexp"
"strconv"
"strings"
)
// Event types corresponding to scm hooks.
const (
EventPush = "push"
EventPull = "pull_request"
EventTag = "tag"
EventDeploy = "deployment"
)
type ( type (
// Metadata defines runtime m. // Metadata defines runtime m.
@ -15,10 +28,11 @@ type (
// Repo defines runtime metadata for a repository. // Repo defines runtime metadata for a repository.
Repo struct { Repo struct {
Name string `json:"name,omitempty"` Name string `json:"name,omitempty"`
Link string `json:"link,omitempty"` Link string `json:"link,omitempty"`
Remote string `json:"remote,omitempty"` Remote string `json:"remote,omitempty"`
Private bool `json:"private,omitempty"` Private bool `json:"private,omitempty"`
Secrets []Secret `json:"secrets,omitempty"`
} }
// Build defines runtime metadata for a build. // Build defines runtime metadata for a build.
@ -59,18 +73,27 @@ type (
Matrix map[string]string `json:"matrix,omitempty"` Matrix map[string]string `json:"matrix,omitempty"`
} }
// Secret defines a runtime secret
Secret struct {
Name string `json:"name,omitempty"`
Value string `json:"value,omitempty"`
Mount string `json:"mount,omitempty"`
Mask bool `json:"mask,omitempty"`
}
// System defines runtime metadata for a ci/cd system. // System defines runtime metadata for a ci/cd system.
System struct { System struct {
Name string `json:"name,omitempty"` Name string `json:"name,omitempty"`
Host string `json:"host,omitempty"` Host string `json:"host,omitempty"`
Link string `json:"link,omitempty"` Link string `json:"link,omitempty"`
Arch string `json:"arch,omitempty"` Arch string `json:"arch,omitempty"`
Version string `json:"version,omitempty"`
} }
) )
// Environ returns the metadata as a map of environment variables. // Environ returns the metadata as a map of environment variables.
func (m *Metadata) Environ() map[string]string { func (m *Metadata) Environ() map[string]string {
return map[string]string{ params := map[string]string{
"CI_REPO": m.Repo.Name, "CI_REPO": m.Repo.Name,
"CI_REPO_NAME": m.Repo.Name, "CI_REPO_NAME": m.Repo.Name,
"CI_REPO_LINK": m.Repo.Link, "CI_REPO_LINK": m.Repo.Link,
@ -116,6 +139,70 @@ func (m *Metadata) Environ() map[string]string {
"CI_SYSTEM_LINK": m.Sys.Link, "CI_SYSTEM_LINK": m.Sys.Link,
"CI_SYSTEM_HOST": m.Sys.Host, "CI_SYSTEM_HOST": m.Sys.Host,
"CI_SYSTEM_ARCH": m.Sys.Arch, "CI_SYSTEM_ARCH": m.Sys.Arch,
"CI_SYSTEM_VERSION": m.Sys.Version,
"CI": m.Sys.Name, "CI": m.Sys.Name,
} }
if m.Curr.Event == EventTag {
params["CI_TAG"] = strings.TrimPrefix(m.Curr.Commit.Ref, "refs/tags/")
}
if m.Curr.Event == EventPull {
params["CI_PULL_REQUEST"] = pullRegexp.FindString(m.Curr.Commit.Ref)
}
return params
} }
// EnvironDrone returns metadata as a map of DRONE_ environment variables.
// This is here for backward compatibility and will eventually be removed.
func (m *Metadata) EnvironDrone() map[string]string {
// MISSING PARAMETERS
// * DRONE_REPO_TRUSTED
// * DRONE_YAML_VERIFIED
// * DRONE_YAML_VERIFIED
params := map[string]string{
"CI": "drone",
"DRONE": "true",
"DRONE_ARCH": "linux/amd64",
"DRONE_REPO": m.Repo.Name,
"DRONE_REPO_SCM": "git",
"DRONE_REPO_OWNER": strings.Split(m.Repo.Name, "/")[0],
"DRONE_REPO_NAME": strings.Split(m.Repo.Name, "/")[0],
"DRONE_REPO_LINK": m.Repo.Link,
"DRONE_REPO_BRANCH": m.Curr.Commit.Branch,
"DRONE_REPO_PRIVATE": fmt.Sprintf("%v", m.Repo.Private),
"DRONE_REPO_TRUSTED": "false", // TODO should this be added?
"DRONE_REMOTE_URL": m.Repo.Remote,
"DRONE_COMMIT_SHA": m.Curr.Commit.Sha,
"DRONE_COMMIT_REF": m.Curr.Commit.Ref,
"DRONE_COMMIT_REFSPEC": m.Curr.Commit.Refspec,
"DRONE_COMMIT_BRANCH": m.Curr.Commit.Branch,
"DRONE_COMMIT_LINK": m.Curr.Link,
"DRONE_COMMIT_MESSAGE": m.Curr.Commit.Message,
"DRONE_COMMIT_AUTHOR": m.Curr.Commit.Author.Name,
"DRONE_COMMIT_AUTHOR_EMAIL": m.Curr.Commit.Author.Email,
"DRONE_COMMIT_AUTHOR_AVATAR": m.Curr.Commit.Author.Avatar,
"DRONE_BUILD_NUMBER": fmt.Sprintf("%d", m.Curr.Number),
"DRONE_BUILD_EVENT": m.Curr.Event,
"DRONE_BUILD_LINK": fmt.Sprintf("%s/%s/%d", m.Sys.Link, m.Repo.Name, m.Curr.Number),
"DRONE_BUILD_CREATED": fmt.Sprintf("%d", m.Curr.Created),
"DRONE_BUILD_STARTED": fmt.Sprintf("%d", m.Curr.Started),
"DRONE_BUILD_FINISHED": fmt.Sprintf("%d", m.Curr.Finished),
"DRONE_JOB_NUMBER": fmt.Sprintf("%d", m.Job.Number),
"DRONE_JOB_STARTED": fmt.Sprintf("%d", m.Curr.Started), // ISSUE: no job started
"DRONE_BRANCH": m.Curr.Commit.Branch,
"DRONE_COMMIT": m.Curr.Commit.Sha,
"DRONE_VERSION": m.Sys.Version,
"DRONE_DEPLOY_TO": m.Curr.Target,
"DRONE_PREV_BUILD_STATUS": m.Prev.Status,
"DRONE_PREV_BUILD_NUMBER": fmt.Sprintf("%v", m.Prev.Number),
"DRONE_PREV_COMMIT_SHA": m.Prev.Commit.Sha,
}
if m.Curr.Event == EventTag {
params["DRONE_TAG"] = strings.TrimPrefix(m.Curr.Commit.Ref, "refs/tags/")
}
if m.Curr.Event == EventPull {
params["DRONE_PULL_REQUEST"] = pullRegexp.FindString(m.Curr.Commit.Ref)
}
return params
}
var pullRegexp = regexp.MustCompile("\\d+")

View file

@ -3,7 +3,6 @@ package compiler
import ( import (
"fmt" "fmt"
"path" "path"
"strings"
"github.com/cncd/pipeline/pipeline/backend" "github.com/cncd/pipeline/pipeline/backend"
"github.com/cncd/pipeline/pipeline/frontend/yaml" "github.com/cncd/pipeline/pipeline/frontend/yaml"
@ -55,12 +54,6 @@ func (c *Compiler) createProcess(name string, container *yaml.Container) *backen
continue continue
default: default:
environment[k] = v environment[k] = v
// legacy code for drone plugins
if strings.HasPrefix(k, "CI_") {
p := strings.Replace(k, "CI_", "DRONE_", 1)
environment[p] = v
}
} }
} }

View file

@ -31,6 +31,11 @@ func WithMetadata(metadata frontend.Metadata) Option {
for k, v := range metadata.Environ() { for k, v := range metadata.Environ() {
compiler.env[k] = v compiler.env[k] = v
} }
// TODO this is present for backward compatibility and should
// be removed in a future version.
for k, v := range metadata.EnvironDrone() {
compiler.env[k] = v
}
} }
} }
@ -42,6 +47,12 @@ func WithNetrc(username, password, machine string) Option {
"CI_NETRC_USERNAME": username, "CI_NETRC_USERNAME": username,
"CI_NETRC_PASSWORD": password, "CI_NETRC_PASSWORD": password,
"CI_NETRC_MACHINE": machine, "CI_NETRC_MACHINE": machine,
// TODO this is present for backward compatibility and should
// be removed in a future version.
"DRONE_NETRC_USERNAME": username,
"DRONE_NETRC_PASSWORD": password,
"DRONE_NETRC_MACHINE": machine,
}, },
) )
} }

View file

@ -8,30 +8,32 @@ import (
"net/textproto" "net/textproto"
) )
// Reader is an iterator over parts in a multipart log stream. type (
type Reader interface { // Reader is an iterator over parts in a multipart log stream.
// NextPart returns the next part in the multipart or Reader interface {
// an error. When there are no more parts, the error // NextPart returns the next part in the multipart or
// io.EOF is returned. // an error. When there are no more parts, the error
NextPart() (Part, error) // io.EOF is returned.
} NextPart() (Part, error)
}
// A Part represents a single part in a multipart body. // A Part represents a single part in a multipart body.
type Part interface { Part interface {
io.Reader io.Reader
// Header returns the headers of the body with the // Header returns the headers of the body with the
// keys canonicalized. // keys canonicalized.
Header() textproto.MIMEHeader Header() textproto.MIMEHeader
// FileName returns the filename parameter of the // FileName returns the filename parameter of the
// Content-Disposition header. // Content-Disposition header.
FileName() string FileName() string
// FormName returns the name parameter if p has a // FormName returns the name parameter if p has a
// Content-Disposition of type form-data. // Content-Disposition of type form-data.
FormName() string FormName() string
} }
)
// New returns a new multipart Reader. // New returns a new multipart Reader.
func New(r io.Reader) Reader { func New(r io.Reader) Reader {
@ -49,7 +51,7 @@ func New(r io.Reader) Reader {
} }
// //
// // wraps the stdlib multi-part reader
// //
type multipartReader struct { type multipartReader struct {
@ -70,7 +72,7 @@ func (r *multipartReader) NextPart() (Part, error) {
} }
// //
// // wraps a simple io.Reader to satisfy the multi-part interface
// //
type textReader struct { type textReader struct {
@ -85,7 +87,6 @@ func (r *textReader) NextPart() (Part, error) {
r.done = true r.done = true
p := new(part) p := new(part)
p.Reader = r.reader p.Reader = r.reader
p.filename = "terminal.log"
return p, nil return p, nil
} }

View file

@ -21,12 +21,12 @@ const (
methodDone = "done" methodDone = "done"
methodExtend = "extend" methodExtend = "extend"
methodUpdate = "update" methodUpdate = "update"
methodSave = "save" methodUpload = "upload"
methodLog = "log" methodLog = "log"
) )
type ( type (
saveReq struct { uploadReq struct {
ID string `json:"id"` ID string `json:"id"`
Mime string `json:"mime"` Mime string `json:"mime"`
Data []byte `json:"data"` Data []byte `json:"data"`
@ -75,9 +75,9 @@ func NewClient(endpoint string, opts ...Option) (*Client, error) {
} }
// Next returns the next pipeline in the queue. // Next returns the next pipeline in the queue.
func (t *Client) Next(c context.Context) (*Pipeline, error) { func (t *Client) Next(c context.Context, f Filter) (*Pipeline, error) {
res := new(Pipeline) res := new(Pipeline)
err := t.call(c, methodNext, nil, res) err := t.call(c, methodNext, f, res)
return res, err return res, err
} }
@ -112,14 +112,14 @@ func (t *Client) Log(c context.Context, id string, line *Line) error {
return t.call(c, methodLog, &params, nil) return t.call(c, methodLog, &params, nil)
} }
// Save saves the pipeline artifact. // Upload uploads the pipeline artifact.
func (t *Client) Save(c context.Context, id, mime string, file io.Reader) error { func (t *Client) Upload(c context.Context, id, mime string, file io.Reader) error {
data, err := ioutil.ReadAll(file) data, err := ioutil.ReadAll(file)
if err != nil { if err != nil {
return err return err
} }
params := saveReq{id, mime, data} params := uploadReq{id, mime, data}
return t.call(c, methodSave, params, nil) return t.call(c, methodUpload, params, nil)
} }
// Close closes the client connection. // Close closes the client connection.

View file

@ -1,9 +1,9 @@
package rpc package rpc
import ( import (
"bytes"
"context" "context"
"fmt" "fmt"
"strings"
"time" "time"
) )
@ -41,30 +41,55 @@ type LineWriter struct {
name string name string
num int num int
now time.Time now time.Time
rep *strings.Replacer
} }
// NewLineWriter returns a new line reader. // NewLineWriter returns a new line reader.
func NewLineWriter(peer Peer, id, name string) *LineWriter { func NewLineWriter(peer Peer, id, name string, secret ...string) *LineWriter {
w := new(LineWriter) w := new(LineWriter)
w.peer = peer w.peer = peer
w.id = id w.id = id
w.name = name w.name = name
w.num = 0 w.num = 0
w.now = time.Now().UTC() w.now = time.Now().UTC()
var oldnew []string
for _, old := range secret {
oldnew = append(oldnew, old)
oldnew = append(oldnew, "********")
}
if len(oldnew) != 0 {
w.rep = strings.NewReplacer(oldnew...)
}
return w return w
} }
func (w *LineWriter) Write(p []byte) (n int, err error) { func (w *LineWriter) Write(p []byte) (n int, err error) {
for _, part := range bytes.Split(p, []byte{'\n'}) { out := string(p)
line := &Line{ if w.rep != nil {
Out: string(part), out = w.rep.Replace(out)
Proc: w.name,
Pos: w.num,
Time: int64(time.Since(w.now).Seconds()),
Type: LineStdout,
}
w.peer.Log(context.Background(), w.id, line)
w.num++
} }
line := &Line{
Out: out,
Proc: w.name,
Pos: w.num,
Time: int64(time.Since(w.now).Seconds()),
Type: LineStdout,
}
w.peer.Log(context.Background(), w.id, line)
w.num++
// for _, part := range bytes.Split(p, []byte{'\n'}) {
// line := &Line{
// Out: string(part),
// Proc: w.name,
// Pos: w.num,
// Time: int64(time.Since(w.now).Seconds()),
// Type: LineStdout,
// }
// w.peer.Log(context.Background(), w.id, line)
// w.num++
// }
return len(p), nil return len(p), nil
} }

View file

@ -13,7 +13,8 @@ import (
type ( type (
// Filter defines filters for fetching items from the queue. // Filter defines filters for fetching items from the queue.
Filter struct { Filter struct {
Platform string `json:"platform"` Labels map[string]string `json:"labels"`
Expr string `json:"expr"`
} }
// State defines the pipeline state. // State defines the pipeline state.
@ -34,10 +35,13 @@ type (
} }
) )
// NoFilter is an empty filter.
var NoFilter = Filter{}
// Peer defines a peer-to-peer connection. // Peer defines a peer-to-peer connection.
type Peer interface { type Peer interface {
// Next returns the next pipeline in the queue. // Next returns the next pipeline in the queue.
Next(c context.Context) (*Pipeline, error) Next(c context.Context, f Filter) (*Pipeline, error)
// Wait blocks untilthe pipeline is complete. // Wait blocks untilthe pipeline is complete.
Wait(c context.Context, id string) error Wait(c context.Context, id string) error
@ -51,9 +55,8 @@ type Peer interface {
// Update updates the pipeline state. // Update updates the pipeline state.
Update(c context.Context, id string, state State) error Update(c context.Context, id string, state State) error
// Save saves the pipeline artifact. // Upload uploads the pipeline artifact.
// TODO rename to Upload Upload(c context.Context, id, mime string, file io.Reader) error
Save(c context.Context, id, mime string, file io.Reader) error
// Log writes the pipeline log entry. // Log writes the pipeline log entry.
Log(c context.Context, id string, line *Line) error Log(c context.Context, id string, line *Line) error

View file

@ -62,8 +62,8 @@ func (s *Server) router(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.
return s.update(req) return s.update(req)
case methodLog: case methodLog:
return s.log(req) return s.log(req)
case methodSave: case methodUpload:
return s.save(req) return s.upload(req)
default: default:
return nil, errNoSuchMethod return nil, errNoSuchMethod
} }
@ -72,7 +72,11 @@ func (s *Server) router(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.
// next unmarshals the rpc request parameters and invokes the peer.Next // next unmarshals the rpc request parameters and invokes the peer.Next
// procedure. The results are retuned and written to the rpc response. // procedure. The results are retuned and written to the rpc response.
func (s *Server) next(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) { func (s *Server) next(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
return s.peer.Next(ctx) in := Filter{}
if err := json.Unmarshal([]byte(*req.Params), &in); err != nil {
return nil, err
}
return s.peer.Next(ctx, in)
} }
// wait unmarshals the rpc request parameters and invokes the peer.Wait // wait unmarshals the rpc request parameters and invokes the peer.Wait
@ -128,10 +132,10 @@ func (s *Server) log(req *jsonrpc2.Request) (interface{}, error) {
return nil, s.peer.Log(noContext, in.ID, in.Line) return nil, s.peer.Log(noContext, in.ID, in.Line)
} }
func (s *Server) save(req *jsonrpc2.Request) (interface{}, error) { func (s *Server) upload(req *jsonrpc2.Request) (interface{}, error) {
in := new(saveReq) in := new(uploadReq)
if err := json.Unmarshal([]byte(*req.Params), in); err != nil { if err := json.Unmarshal([]byte(*req.Params), in); err != nil {
return nil, err return nil, err
} }
return nil, s.peer.Save(noContext, in.ID, in.Mime, bytes.NewBuffer(in.Data)) return nil, s.peer.Upload(noContext, in.ID, in.Mime, bytes.NewBuffer(in.Data))
} }

54
vendor/vendor.json vendored
View file

@ -33,68 +33,68 @@
{ {
"checksumSHA1": "W3AuK8ocqHwlUajGmQLFvnRhTZE=", "checksumSHA1": "W3AuK8ocqHwlUajGmQLFvnRhTZE=",
"path": "github.com/cncd/pipeline/pipeline", "path": "github.com/cncd/pipeline/pipeline",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "PSzh0ix/rlMrS/Cl3aH6GHGrJuo=", "checksumSHA1": "Qu2FreqaMr8Yx2bW9O0cxAGgjr0=",
"path": "github.com/cncd/pipeline/pipeline/backend", "path": "github.com/cncd/pipeline/pipeline/backend",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "0CGXRaYwZhJxGIrGhn8WGpkFqPo=", "checksumSHA1": "0CGXRaYwZhJxGIrGhn8WGpkFqPo=",
"path": "github.com/cncd/pipeline/pipeline/backend/docker", "path": "github.com/cncd/pipeline/pipeline/backend/docker",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "uUagpzha5ah/a3RO6IImvzHYFlY=", "checksumSHA1": "/8wE+cVb7T4PQZgpLNu0DHzKGuE=",
"path": "github.com/cncd/pipeline/pipeline/frontend", "path": "github.com/cncd/pipeline/pipeline/frontend",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "O0sulBQAHJeNLg3lO38Cq5uf/eg=", "checksumSHA1": "O0sulBQAHJeNLg3lO38Cq5uf/eg=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "+4c/I/PEDCgzog8m4ohw1parhgE=", "checksumSHA1": "Iu+QmUqkN9ZsBdmVlCclVKthJbM=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/compiler", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/compiler",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "Q0GkNUFamVYIA1Fd8r0A5M6Gx54=", "checksumSHA1": "Q0GkNUFamVYIA1Fd8r0A5M6Gx54=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/linter", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/linter",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "kx2sPUIMozPC/g6E4w48h3FfH3k=", "checksumSHA1": "kx2sPUIMozPC/g6E4w48h3FfH3k=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/matrix", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/matrix",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "2/3f3oNmxXy5kcrRLCFa24Oc9O4=", "checksumSHA1": "2/3f3oNmxXy5kcrRLCFa24Oc9O4=",
"path": "github.com/cncd/pipeline/pipeline/interrupt", "path": "github.com/cncd/pipeline/pipeline/interrupt",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "8eTwXZPM/Kp9uE/mnhpWDTiX7nY=", "checksumSHA1": "/nHBigDoEi2F6zJzvCWOvJ3um2c=",
"path": "github.com/cncd/pipeline/pipeline/multipart", "path": "github.com/cncd/pipeline/pipeline/multipart",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "UUmeGDBdpk+UXtexFnNmbWIHgG8=", "checksumSHA1": "MratmNKJ78/IhWvDsZphN01CtmE=",
"path": "github.com/cncd/pipeline/pipeline/rpc", "path": "github.com/cncd/pipeline/pipeline/rpc",
"revision": "d4e09fd3021a16408bc3ebdd3500efd28f51e72c", "revision": "687ea03140263b4774505c44f212dd4999faa534",
"revisionTime": "2017-03-05T09:53:47Z" "revisionTime": "2017-03-12T08:45:42Z"
}, },
{ {
"checksumSHA1": "7Qj1DK0ceAXkYztW0l3+L6sn+V8=", "checksumSHA1": "7Qj1DK0ceAXkYztW0l3+L6sn+V8=",