initial work to enable grpc

.drone.yml

@@ -46,7 +46,7 @@ pipeline:
     image: plugins/docker
     repo: drone/drone
     secrets: [ docker_username, docker_password ]
-    tag: [ 0.7, 0.7.3 ]
+    tag: [ 0.8, 0.8.0, 0.8.0-rc.1 ]
     when:
       event: tag
 

Dockerfile

@@ -1,7 +1,7 @@
 # docker build --rm -t drone/drone .
 
 FROM centurylink/ca-certs
-EXPOSE 8000 80 443
+EXPOSE 8000 9000 80 443
 
 ENV DATABASE_DRIVER=sqlite3
 ENV DATABASE_CONFIG=/var/lib/drone/drone.sqlite

drone/agent/agent.go

@@ -6,19 +6,18 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
-	"math"
-	"net/url"
 	"strconv"
 	"sync"
 	"time"
 
+	"google.golang.org/grpc"
+
 	"github.com/cncd/pipeline/pipeline"
 	"github.com/cncd/pipeline/pipeline/backend"
 	"github.com/cncd/pipeline/pipeline/backend/docker"
 	"github.com/cncd/pipeline/pipeline/interrupt"
 	"github.com/cncd/pipeline/pipeline/multipart"
 	"github.com/cncd/pipeline/pipeline/rpc"
-	"github.com/drone/drone/version"
 
 	"github.com/tevino/abool"
 	"github.com/urfave/cli"
@@ -31,83 +30,60 @@ var Command = cli.Command{
 	Action: loop,
 	Flags: []cli.Flag{
 		cli.StringFlag{
-			EnvVar: "DRONE_SERVER,DRONE_ENDPOINT",
-			Name:   "drone-server",
+			EnvVar: "DRONE_SERVER",
+			Name:   "server",
 			Usage:  "drone server address",
-			Value:  "ws://localhost:8000/ws/broker",
+			Value:  "localhost:9000",
 		},
 		cli.StringFlag{
-			EnvVar: "DRONE_SECRET,DRONE_AGENT_SECRET",
-			Name:   "drone-secret",
+			EnvVar: "DRONE_SECRET",
+			Name:   "secret",
 			Usage:  "drone agent secret",
 		},
-		cli.DurationFlag{
-			EnvVar: "DRONE_BACKOFF",
-			Name:   "backoff",
-			Usage:  "drone server backoff interval",
-			Value:  time.Second * 15,
-		},
-		cli.IntFlag{
-			Name:   "retry-limit",
-			EnvVar: "DRONE_RETRY_LIMIT",
-			Value:  math.MaxInt32,
-		},
 		cli.BoolFlag{
 			EnvVar: "DRONE_DEBUG",
 			Name:   "debug",
 			Usage:  "start the agent in debug mode",
 		},
 		cli.StringFlag{
-			EnvVar: "DRONE_FILTER",
-			Name:   "filter",
-			Usage:  "filter jobs processed by this agent",
+			EnvVar: "DRONE_PLATFORM",
+			Name:   "platform",
+			Value:  "linux/amd64",
 		},
 		cli.IntFlag{
-			Name:   "max-procs",
 			EnvVar: "DRONE_MAX_PROCS",
+			Name:   "max-procs",
 			Value:  1,
 		},
-		cli.StringFlag{
-			Name:   "platform",
-			EnvVar: "DRONE_PLATFORM",
-			Value:  "linux/amd64",
-		},
 	},
 }
 
 func loop(c *cli.Context) error {
-	endpoint, err := url.Parse(
-		c.String("drone-server"),
-	)
-	if err != nil {
-		return err
-	}
+	// endpoint, err := url.Parse(
+	// 	c.String("drone-server"),
+	// )
+	// if err != nil {
+	// 	return err
+	// }
 	filter := rpc.Filter{
 		Labels: map[string]string{
 			"platform": c.String("platform"),
 		},
 	}
 
-	client, err := rpc.NewClient(
-		endpoint.String(),
-		rpc.WithRetryLimit(
-			c.Int("retry-limit"),
-		),
-		rpc.WithBackoff(
-			c.Duration("backoff"),
-		),
-		rpc.WithToken(
-			c.String("drone-secret"),
-		),
-		rpc.WithHeader(
-			"X-Drone-Version",
-			version.Version.String(),
-		),
+	// TODO pass version information to grpc server
+	// TODO authenticate to grpc server
+
+	conn, err := grpc.Dial(
+		c.String("server"),
+		grpc.WithInsecure(),
 	)
 	if err != nil {
 		return err
 	}
-	defer client.Close()
+	defer conn.Close()
+
+	client := rpc.NewGrpcClient(conn)
 
 	sigterm := abool.New()
 	ctx := context.Background()

drone/server/server.go

@@ -2,19 +2,24 @@ package server
 
 import (
 	"context"
+	"net"
 	"net/http"
 	"net/url"
 	"strings"
 	"time"
 
+	"google.golang.org/grpc"
+
 	"golang.org/x/crypto/acme/autocert"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/cncd/logging"
+	"github.com/cncd/pipeline/pipeline/rpc/proto"
 	"github.com/cncd/pubsub"
 	"github.com/drone/drone/plugins/registry"
 	"github.com/drone/drone/plugins/secrets"
 	"github.com/drone/drone/plugins/sender"
+	"github.com/drone/drone/remote"
 	"github.com/drone/drone/router"
 	"github.com/drone/drone/router/middleware"
 	droneserver "github.com/drone/drone/server"
@@ -395,19 +400,52 @@ func server(c *cli.Context) error {
 		logrus.SetLevel(logrus.WarnLevel)
 	}
 
-	s := setupStore(c)
-	setupEvilGlobals(c, s)
+	remote_, err := setupRemote(c)
+	if err != nil {
+		logrus.Fatal(err)
+	}
+
+	store_ := setupStore(c)
+	setupEvilGlobals(c, store_, remote_)
 
 	// setup the server and start the listener
 	handler := router.Load(
 		ginrus.Ginrus(logrus.StandardLogger(), time.RFC3339, true),
 		middleware.Version,
 		middleware.Config(c),
 		middleware.Cache(c),
-		middleware.Store(c, s),
-		middleware.Remote(c),
+		middleware.Store(c, store_),
+		middleware.Remote(remote_),
 	)
 
+	var g errgroup.Group
+
+	// start the grpc server
+	g.Go(func() error {
+
+		lis, err := net.Listen("tcp", ":9000")
+		if err != nil {
+			logrus.Error(err)
+			return err
+		}
+		s := grpc.NewServer()
+		ss := new(droneserver.DroneServer)
+		ss.Queue = droneserver.Config.Services.Queue
+		ss.Logger = droneserver.Config.Services.Logs
+		ss.Pubsub = droneserver.Config.Services.Pubsub
+		ss.Remote = remote_
+		ss.Store = store_
+		ss.Host = droneserver.Config.Server.Host
+		proto.RegisterDroneServer(s, ss)
+
+		err = s.Serve(lis)
+		if err != nil {
+			logrus.Error(err)
+			return err
+		}
+		return nil
+	})
+
 	// start the server with tls enabled
 	if c.String("server-cert") != "" {
 		return http.ListenAndServeTLS(
@@ -428,7 +466,6 @@ func server(c *cli.Context) error {
 
 	// start the server with lets encrypt enabled
 	// listen on ports 443 and 80
-	var g errgroup.Group
 	g.Go(func() error {
 		return http.ListenAndServe(":http", handler)
 	})
@@ -449,7 +486,7 @@ func server(c *cli.Context) error {
 // in the gin.Context to storing them in a struct. We are also moving away
 // from gin to gorilla. We will temporarily use global during our refactoring
 // which will be removing in the final implementation.
-func setupEvilGlobals(c *cli.Context, v store.Store) {
+func setupEvilGlobals(c *cli.Context, v store.Store, r remote.Remote) {
 
 	// storage
 	droneserver.Config.Storage.Files = v
@@ -463,6 +500,7 @@ func setupEvilGlobals(c *cli.Context, v store.Store) {
 	droneserver.Config.Services.Registries = setupRegistryService(c, v)
 	droneserver.Config.Services.Secrets = setupSecretService(c, v)
 	droneserver.Config.Services.Senders = sender.New(v, v)
+
 	if endpoint := c.String("registry-service"); endpoint != "" {
 		droneserver.Config.Services.Registries = registry.NewRemote(endpoint)
 	}

drone/server/setup.go

@@ -3,10 +3,19 @@
 package server
 
 import (
+	"fmt"
+
 	"github.com/cncd/queue"
 	"github.com/drone/drone/model"
 	"github.com/drone/drone/plugins/registry"
 	"github.com/drone/drone/plugins/secrets"
+	"github.com/drone/drone/remote"
+	"github.com/drone/drone/remote/bitbucket"
+	"github.com/drone/drone/remote/bitbucketserver"
+	"github.com/drone/drone/remote/gitea"
+	"github.com/drone/drone/remote/github"
+	"github.com/drone/drone/remote/gitlab"
+	"github.com/drone/drone/remote/gogs"
 	"github.com/drone/drone/store"
 	"github.com/drone/drone/store/datastore"
 
@@ -35,3 +44,95 @@ func setupRegistryService(c *cli.Context, s store.Store) model.RegistryService {
 func setupPubsub(c *cli.Context)        {}
 func setupStream(c *cli.Command)        {}
 func setupGatingService(c *cli.Command) {}
+
+// helper function to setup the remote from the CLI arguments.
+func setupRemote(c *cli.Context) (remote.Remote, error) {
+	switch {
+	case c.Bool("github"):
+		return setupGithub(c)
+	case c.Bool("gitlab"):
+		return setupGitlab(c)
+	case c.Bool("bitbucket"):
+		return setupBitbucket(c)
+	case c.Bool("stash"):
+		return setupStash(c)
+	case c.Bool("gogs"):
+		return setupGogs(c)
+	case c.Bool("gitea"):
+		return setupGitea(c)
+	default:
+		return nil, fmt.Errorf("version control system not configured")
+	}
+}
+
+// helper function to setup the Bitbucket remote from the CLI arguments.
+func setupBitbucket(c *cli.Context) (remote.Remote, error) {
+	return bitbucket.New(
+		c.String("bitbucket-client"),
+		c.String("bitbucket-secret"),
+	), nil
+}
+
+// helper function to setup the Gogs remote from the CLI arguments.
+func setupGogs(c *cli.Context) (remote.Remote, error) {
+	return gogs.New(gogs.Opts{
+		URL:         c.String("gogs-server"),
+		Username:    c.String("gogs-git-username"),
+		Password:    c.String("gogs-git-password"),
+		PrivateMode: c.Bool("gogs-private-mode"),
+		SkipVerify:  c.Bool("gogs-skip-verify"),
+	})
+}
+
+// helper function to setup the Gitea remote from the CLI arguments.
+func setupGitea(c *cli.Context) (remote.Remote, error) {
+	return gitea.New(gitea.Opts{
+		URL:         c.String("gitea-server"),
+		Username:    c.String("gitea-git-username"),
+		Password:    c.String("gitea-git-password"),
+		PrivateMode: c.Bool("gitea-private-mode"),
+		SkipVerify:  c.Bool("gitea-skip-verify"),
+	})
+}
+
+// helper function to setup the Stash remote from the CLI arguments.
+func setupStash(c *cli.Context) (remote.Remote, error) {
+	return bitbucketserver.New(bitbucketserver.Opts{
+		URL:               c.String("stash-server"),
+		Username:          c.String("stash-git-username"),
+		Password:          c.String("stash-git-password"),
+		ConsumerKey:       c.String("stash-consumer-key"),
+		ConsumerRSA:       c.String("stash-consumer-rsa"),
+		ConsumerRSAString: c.String("stash-consumer-rsa-string"),
+		SkipVerify:        c.Bool("stash-skip-verify"),
+	})
+}
+
+// helper function to setup the Gitlab remote from the CLI arguments.
+func setupGitlab(c *cli.Context) (remote.Remote, error) {
+	return gitlab.New(gitlab.Opts{
+		URL:         c.String("gitlab-server"),
+		Client:      c.String("gitlab-client"),
+		Secret:      c.String("gitlab-secret"),
+		Username:    c.String("gitlab-git-username"),
+		Password:    c.String("gitlab-git-password"),
+		PrivateMode: c.Bool("gitlab-private-mode"),
+		SkipVerify:  c.Bool("gitlab-skip-verify"),
+	})
+}
+
+// helper function to setup the GitHub remote from the CLI arguments.
+func setupGithub(c *cli.Context) (remote.Remote, error) {
+	return github.New(github.Opts{
+		URL:         c.String("github-server"),
+		Context:     c.String("github-context"),
+		Client:      c.String("github-client"),
+		Secret:      c.String("github-secret"),
+		Scopes:      c.StringSlice("github-scope"),
+		Username:    c.String("github-git-username"),
+		Password:    c.String("github-git-password"),
+		PrivateMode: c.Bool("github-private-mode"),
+		SkipVerify:  c.Bool("github-skip-verify"),
+		MergeRef:    c.BoolT("github-merge-ref"),
+	})
+}

model/queue.go

@@ -62,7 +62,7 @@ func (q *persistentQueue) Poll(c context.Context, f queue.Filter) (*queue.Task,
 		if derr := q.store.TaskDelete(task.ID); derr != nil {
 			logrus.Errorf("pull queue item: %s: failed to remove from backup: %s", task.ID, derr)
 		} else {
-			logrus.Errorf("pull queue item: %s: successfully removed from backup", task.ID)
+			logrus.Debugf("pull queue item: %s: successfully removed from backup", task.ID)
 		}
 	}
 	return task, err