Support redeployability

We create a configuration called rootURL that specificies the root URL
of the taskcluster service.

engines/docker/engine.go

@@ -60,7 +60,7 @@ func (p engineProvider) NewEngine(options engines.EngineOptions) (engines.Engine
 		Environment: env,
 		monitor:     monitor,
 		networks:    network.NewPool(client, monitor.WithPrefix("network-pool")),
-		imageCache:  imagecache.New(client, env.GarbageCollector, monitor.WithPrefix("image-cache")),
+		imageCache:  imagecache.New(client, env, monitor.WithPrefix("image-cache")),
 	}, nil
 }
 

engines/docker/imagecache/context.go

@@ -4,6 +4,7 @@ package imagecache
 
 import (
 	"fmt"
+	"net/url"
 
 	"github.com/taskcluster/taskcluster-worker/runtime"
 	"github.com/taskcluster/taskcluster-worker/runtime/caching"
@@ -14,20 +15,30 @@ import (
 // to match the interface of fetcher.Context
 type cachingContextWithQueue struct {
 	caching.Context
-	queue func() client.Queue
+	queue   func() client.Queue
+	rootURL *url.URL
 }
 
 func (c cachingContextWithQueue) Queue() client.Queue {
 	return c.queue()
 }
 
+func (c cachingContextWithQueue) RootURL() *url.URL {
+	return c.rootURL
+}
+
 // taskContextWithProgress wraps TaskContext to satisfy the caching.Context
 // interface, by adding a Progress() function
 type taskContextWithProgress struct {
 	*runtime.TaskContext
-	Prefix string
+	Prefix  string
+	rootURL *url.URL
 }
 
 func (c taskContextWithProgress) Progress(description string, percent float64) {
 	c.Log(fmt.Sprintf("%s: %s %.0f %%", c.Prefix, description, percent*100))
 }
+
+func (c taskContextWithProgress) RootURL() *url.URL {
+	return c.rootURL
+}

engines/docker/imagecache/imagecache.go

@@ -18,7 +18,6 @@ import (
 	"github.com/taskcluster/taskcluster-worker/runtime/caching"
 	"github.com/taskcluster/taskcluster-worker/runtime/client"
 	"github.com/taskcluster/taskcluster-worker/runtime/fetcher"
-	"github.com/taskcluster/taskcluster-worker/runtime/gc"
 	"github.com/taskcluster/taskcluster-worker/runtime/util"
 )
 
@@ -28,15 +27,17 @@ type ImageCache struct {
 	cache   *caching.Cache
 	docker  *docker.Client
 	monitor runtime.Monitor
+	env     *runtime.Environment
 }
 
 // New creates a new ImageCache object
-func New(d *docker.Client, tracker gc.ResourceTracker, monitor runtime.Monitor) *ImageCache {
+func New(d *docker.Client, env *runtime.Environment, monitor runtime.Monitor) *ImageCache {
 	ic := &ImageCache{
 		docker:  d,
 		monitor: monitor,
+		env:     env,
 	}
-	ic.cache = caching.New(ic.constructor, true, tracker, monitor)
+	ic.cache = caching.New(ic.constructor, true, env.GarbageCollector, monitor)
 	return ic
 }
 
@@ -83,7 +84,7 @@ func (ic *ImageCache) constructor(ctx caching.Context, opts interface{}) (cachin
 	}
 
 	// Load from reference
-	return ic.dockerLoadFromReference(cachingContextWithQueue{ctx, options.queue}, options.reference)
+	return ic.dockerLoadFromReference(cachingContextWithQueue{ctx, options.queue, ic.env.RootURL}, options.reference)
 }
 
 // ImageHandle wraps caching.Handle such that we don't need to do any casting
@@ -109,7 +110,7 @@ func (ic *ImageCache) Require(ctx *runtime.TaskContext, imagePayload interface{}
 		options.Image = s
 	} else {
 		prefix = "Fetching image"
-		ref, err := imageFetcher.NewReference(taskContextWithProgress{ctx, "Resolving image reference"}, imagePayload)
+		ref, err := imageFetcher.NewReference(taskContextWithProgress{ctx, "Resolving image reference", ic.env.RootURL}, imagePayload)
 		if err != nil {
 			return nil, runtime.NewMalformedPayloadError(fmt.Sprintf(
 				"unable to resolve docker image reference error: %s", err.Error(),
@@ -132,7 +133,7 @@ func (ic *ImageCache) Require(ctx *runtime.TaskContext, imagePayload interface{}
 		options.queue = ctx.Queue
 	}
 
-	handle, err := ic.cache.Require(taskContextWithProgress{ctx, prefix}, options)
+	handle, err := ic.cache.Require(taskContextWithProgress{ctx, prefix, ic.env.RootURL}, options)
 	if err != nil {
 		return nil, err
 	}

engines/qemu/fetcher.go

@@ -2,6 +2,7 @@ package qemuengine
 
 import (
 	"fmt"
+	"net/url"
 
 	"github.com/taskcluster/taskcluster-worker/runtime"
 	"github.com/taskcluster/taskcluster-worker/runtime/fetcher"
@@ -21,8 +22,13 @@ var imageFetcher = fetcher.Combine(
 
 type fetchImageContext struct {
 	*runtime.TaskContext
+	rootURL *url.URL
 }
 
 func (c fetchImageContext) Progress(description string, percent float64) {
 	c.Log(fmt.Sprintf("Fetching image: %s - %.0f %%", description, percent*100))
 }
+
+func (c fetchImageContext) RootURL() *url.URL {
+	return c.rootURL
+}

engines/qemu/sandboxbuilder.go

@@ -58,7 +58,7 @@ func newSandboxBuilder(
 		var scopeSets [][]string
 		var inst *image.Instance
 
-		ctx := &fetchImageContext{c}
+		ctx := &fetchImageContext{c, e.Environment.RootURL}
 		ref, err := imageFetcher.NewReference(ctx, payload.Image)
 		if err != nil {
 			goto handleErr

examples/macosx-config.yml

@@ -31,7 +31,6 @@ config:
         TERM_PROGRAM: {$env: TERM_PROGRAM}
         SHELL: /bin/bash
   pollingInterval:  10
-  queueBaseUrl:     https://queue.taskcluster.net/v1
   reclaimOffset:    120
   temporaryFolder:  /tmp/tc-worker-tmp
   serverIp:           127.0.0.1

examples/mock-config.yml

@@ -15,7 +15,6 @@ logLevel:         debug
 plugins:
   disabled:       []
 pollingInterval:  1
-queueBaseUrl:     https://queue.taskcluster.net/v1
 reclaimOffset:    120
 temporaryFolder:  /tmp/tc-worker-tmp
 serverIp:           127.0.0.1

examples/script-config.yml

@@ -39,7 +39,6 @@ config:
       - artifacts
       - env
   pollingInterval:  5
-  queueBaseUrl:     https://queue.taskcluster.net/v1
   reclaimOffset:    120
   temporaryFolder:  /tmp/tc-worker/
   serverIp:         127.0.0.1

plugins/cache/cache.go

@@ -120,7 +120,7 @@ func (p *plugin) getVolume(ctx *runtime.TaskContext, options payloadEntry) (*cac
 	}
 
 	// Create a context with progress reporting capability
-	progressCtx := progressContext{ctx, options.Name}
+	progressCtx := progressContext{ctx, options.Name, p.environment.RootURL}
 
 	// First we resolve the reference, if there is any
 	var ref fetcher.Reference

plugins/cache/constructor.go

@@ -3,6 +3,7 @@ package cache
 import (
 	"fmt"
 	"io"
+	"net/url"
 	"time"
 
 	"github.com/pkg/errors"
@@ -26,15 +27,21 @@ type cacheOptions struct {
 type preloadFetchContext struct {
 	caching.Context
 	InitialTaskContext *runtime.TaskContext
+	rootURL            *url.URL
 }
 
 func (c *preloadFetchContext) Queue() client.Queue {
 	return c.InitialTaskContext.Queue()
 }
 
+func (c *preloadFetchContext) RootURL() *url.URL {
+	return c.rootURL
+}
+
 type progressContext struct {
 	*runtime.TaskContext
-	Name string
+	Name    string
+	rootURL *url.URL
 }
 
 func (c *progressContext) Progress(description string, percent float64) {
@@ -45,6 +52,10 @@ func (c *progressContext) Progress(description string, percent float64) {
 	}
 }
 
+func (c *progressContext) RootURL() *url.URL {
+	return c.rootURL
+}
+
 func constructor(ctx caching.Context, opts interface{}) (caching.Resource, error) {
 	options := opts.(cacheOptions) // must of this type
 

plugins/interactive/interactive.go

@@ -21,14 +21,6 @@ import (
 // configured or given in the task definition
 const defaultArtifactPrefix = "private/interactive/"
 
-// defaultShellToolURL is the default URL for the tool that can connect to the
-// shell socket and display an interactive shell session.
-const defaultShellToolURL = "https://tools.taskcluster.net/shell/"
-
-// defaultShellToolURL is the default URL for the tool that can list displays
-// and connect to the display socket with interactive noVNC session.
-const defaultDisplayToolURL = "https://tools.taskcluster.net/display/"
-
 type provider struct {
 	plugins.PluginProviderBase
 }
@@ -39,15 +31,16 @@ func (provider) ConfigSchema() schematypes.Schema {
 func (provider) NewPlugin(options plugins.PluginOptions) (plugins.Plugin, error) {
 	var c config
 	schematypes.MustValidateAndMap(configSchema, options.Config, &c)
+	toolsURL := options.Environment.GetServiceURL("tools")
 
 	if c.ArtifactPrefix == "" {
 		c.ArtifactPrefix = defaultArtifactPrefix
 	}
 	if c.ShellToolURL == "" {
-		c.ShellToolURL = defaultShellToolURL
+		c.ShellToolURL = fmt.Sprintf("%s/shell", toolsURL)
 	}
 	if c.DisplayToolURL == "" {
-		c.DisplayToolURL = defaultDisplayToolURL
+		c.DisplayToolURL = fmt.Sprintf("%s/display", toolsURL)
 	}
 
 	// IF no WebHookServer is available we disabling the interactive plugin

plugins/livelog/livelog.go

@@ -202,7 +202,12 @@ func (tp *taskPlugin) uploadLog() error {
 		return err // Upload error isn't fatal
 	}
 
-	backingURL := fmt.Sprintf("https://queue.taskcluster.net/v1/task/%s/runs/%d/artifacts/public/logs/live_backing.log", tp.context.TaskInfo.TaskID, tp.context.TaskInfo.RunID)
+	backingURL := fmt.Sprintf(
+		"%s/v1/task/%s/runs/%d/artifacts/public/logs/live_backing.log",
+		tp.environment.GetServiceURL("queue"),
+		tp.context.TaskInfo.TaskID,
+		tp.context.TaskInfo.RunID,
+	)
 	err = tp.context.CreateRedirectArtifact(runtime.RedirectArtifact{
 		Name:     "public/logs/live.log",
 		Mimetype: "text/plain; charset=utf-8",

runtime/environment.go

@@ -1,6 +1,8 @@
 package runtime
 
 import (
+	"net/url"
+
 	"github.com/taskcluster/taskcluster-worker/runtime/gc"
 	"github.com/taskcluster/taskcluster-worker/runtime/webhookserver"
 )
@@ -19,4 +21,21 @@ type Environment struct {
 	WorkerType    string
 	WorkerGroup   string
 	WorkerID      string
+	RootURL       *url.URL
+}
+
+// GetServiceURL takes a service name and returns the full taskcluster
+// URL of it.
+func GetServiceURL(rootURL *url.URL, serviceName string) string {
+	copyURL, err := url.Parse(rootURL.String())
+	if err != nil {
+		panic("WAT???")
+	}
+	copyURL.Host = serviceName + "." + copyURL.Host
+	return copyURL.String()
+}
+
+// GetServiceURL returns the URL of the given service
+func (env *Environment) GetServiceURL(serviceName string) string {
+	return GetServiceURL(env.RootURL, serviceName)
 }

runtime/fetcher/artifactfetcher.go

@@ -10,6 +10,7 @@ import (
 	"github.com/pkg/errors"
 	schematypes "github.com/taskcluster/go-schematypes"
 	"github.com/taskcluster/httpbackoff"
+	"github.com/taskcluster/taskcluster-worker/runtime"
 	"github.com/taskcluster/taskcluster-worker/runtime/util"
 )
 
@@ -95,8 +96,7 @@ func (r *artifactReference) Fetch(ctx Context, target WriteReseter) error {
 	// Construct URL
 	var u string
 	if r.isPublic() {
-		// TODO: Get queueBaseUrl from TaskContext somehow... and facilitate mocking in tests
-		u = fmt.Sprintf("https://queue.taskcluster.net/v1/task/%s/runs/%d/artifacts/%s", r.TaskID, r.RunID, r.Artifact)
+		u = fmt.Sprintf("%s/v1/task/%s/runs/%d/artifacts/%s", runtime.GetServiceURL(ctx.RootURL(), "queue"), r.TaskID, r.RunID, r.Artifact)
 	} else {
 		u2, err := ctx.Queue().GetArtifact_SignedURL(r.TaskID, strconv.Itoa(r.RunID), r.Artifact, 25*time.Minute)
 		if err != nil {

runtime/fetcher/context.go

@@ -2,6 +2,7 @@ package fetcher
 
 import (
 	"context"
+	"net/url"
 	"time"
 
 	"github.com/taskcluster/taskcluster-worker/runtime/client"
@@ -21,11 +22,13 @@ type Context interface {
 	// consumers may wish to round to one decimal using "%.0f" formatting.
 	// Progress reports won't be sent more than once every 5 seconds.
 	Progress(description string, percent float64)
+	RootURL() *url.URL
 }
 
 type contextWithCancel struct {
 	context.Context
-	parent Context
+	parent  Context
+	rootURL *url.URL
 }
 
 func (c *contextWithCancel) Queue() client.Queue {
@@ -36,8 +39,12 @@ func (c *contextWithCancel) Progress(description string, percent float64) {
 	c.parent.Progress(description, percent)
 }
 
+func (c *contextWithCancel) RootURL() *url.URL {
+	return c.rootURL
+}
+
 // WithCancel returns a Context and a cancel function similar to context.WithCancel
-func WithCancel(ctx Context) (Context, func()) {
+func WithCancel(ctx Context, rootURL *url.URL) (Context, func()) {
 	child, cancel := context.WithCancel(ctx)
-	return &contextWithCancel{child, ctx}, cancel
+	return &contextWithCancel{child, ctx, rootURL}, cancel
 }

test/testdata/worker-config.yml

@@ -21,7 +21,6 @@ config:
       groups: []
   logLevel:         debug
   pollingInterval:  1
-  queueBaseUrl:     https://queue.taskcluster.net/v1
   reclaimOffset:    120
   temporaryFolder:  /var/tmp/tc-worker-tmp
   serverIp:           127.0.0.1

worker/config.go

@@ -34,8 +34,7 @@ type configType struct {
 	MinimumMemory    int64                  `json:"minimumMemory"`
 	Monitor          interface{}            `json:"monitor"`
 	Credentials      tcclient.Credentials   `json:"credentials"`
-	QueueBaseURL     string                 `json:"queueBaseUrl"`
-	AuthBaseURL      string                 `json:"authBaseUrl"`
+	RootURL          string                 `json:"rootURL"`
 	WorkerOptions    options                `json:"worker"`
 }
 
@@ -224,11 +223,10 @@ func ConfigSchema() schematypes.Object {
 				Minimum: 0,
 				Maximum: math.MaxInt64,
 			},
-			"monitor":      monitoring.ConfigSchema,
-			"credentials":  credentialsSchema,
-			"queueBaseUrl": schematypes.String{},
-			"authBaseUrl":  schematypes.String{},
-			"worker":       optionsSchema,
+			"rootURL":     schematypes.String{},
+			"monitor":     monitoring.ConfigSchema,
+			"credentials": credentialsSchema,
+			"worker":      optionsSchema,
 		},
 		Required: []string{
 			"engine",