The first implementation of the reboot functionallity worked
under the assumption that we use file based hook locks. This meant
that the machine agent could acquire a lock and choose not to break
it until it was restarted, thus blocking any unit agents from running
hooks.

This decission was made under the pretext that the unit agent would
soon be merged with the machine agent, and switching from file locks
to mutexes/semaphores would not break it.

Juju has moved away from file locks, and we can no longer guarantee
a lock will be held by the machine agent until the machine is fully
rebooted. On reboot/shutdown the init system can now stop the machine
agent first, allowing unit agents to start executing hooks.

This change stops all deployed juju unit agents prior to issuing the
reboot (or shutdown) command.

api/apiclient.go

@@ -44,6 +44,9 @@ const PingPeriod = 1 * time.Minute
 // consider it to have failed.
 const pingTimeout = 30 * time.Second
 
+// modelRoot is the prefix that all model API paths begin with.
+const modelRoot = "/model/"
+
 var logger = loggo.GetLogger("juju.api")
 
 type rpcConnection interface {
@@ -277,6 +280,34 @@ func (t *hostSwitchingTransport) RoundTrip(req *http.Request) (*http.Response, e
 
 // ConnectStream implements StreamConnector.ConnectStream.
 func (st *state) ConnectStream(path string, attrs url.Values) (base.Stream, error) {
+	path, err := apiPath(st.modelTag, path)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	conn, err := st.connectStreamWithRetry(path, attrs, nil)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	return conn, nil
+}
+
+// ConnectControllerStream creates a stream connection to an API path
+// that isn't prefixed with /model/uuid.
+func (st *state) ConnectControllerStream(path string, attrs url.Values, headers http.Header) (base.Stream, error) {
+	if !strings.HasPrefix(path, "/") {
+		return nil, errors.Errorf("path %q is not absolute", path)
+	}
+	if strings.HasPrefix(path, modelRoot) {
+		return nil, errors.Errorf("path %q is model-specific", path)
+	}
+	conn, err := st.connectStreamWithRetry(path, attrs, headers)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	return conn, nil
+}
+
+func (st *state) connectStreamWithRetry(path string, attrs url.Values, headers http.Header) (base.Stream, error) {
 	if !st.isLoggedIn() {
 		return nil, errors.New("cannot use ConnectStream without logging in")
 	}
@@ -286,7 +317,7 @@ func (st *state) ConnectStream(path string, attrs url.Values) (base.Stream, erro
 	// error, the response will contain a macaroon that, when discharged,
 	// may allow access, so we discharge it (using bakery.Client.HandleError)
 	// and try the request again.
-	conn, err := st.connectStream(path, attrs)
+	conn, err := st.connectStream(path, attrs, headers)
 	if err == nil {
 		return conn, err
 	}
@@ -297,7 +328,7 @@ func (st *state) ConnectStream(path string, attrs url.Values) (base.Stream, erro
 		return nil, errors.Trace(err)
 	}
 	// Try again with the discharged macaroon.
-	conn, err = st.connectStream(path, attrs)
+	conn, err = st.connectStream(path, attrs, headers)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -307,11 +338,7 @@ func (st *state) ConnectStream(path string, attrs url.Values) (base.Stream, erro
 // connectStream is the internal version of ConnectStream. It differs from
 // ConnectStream only in that it will not retry the connection if it encounters
 // discharge-required error.
-func (st *state) connectStream(path string, attrs url.Values) (base.Stream, error) {
-	path, err := apiPath(st.modelTag, path)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
+func (st *state) connectStream(path string, attrs url.Values, extraHeaders http.Header) (base.Stream, error) {
 	target := url.URL{
 		Scheme:   "wss",
 		Host:     st.addr,
@@ -331,6 +358,11 @@ func (st *state) connectStream(path string, attrs url.Values) (base.Stream, erro
 	// Add any cookies because they will not be sent to websocket
 	// connections by default.
 	st.addCookiesToHeader(cfg.Header)
+	for header, values := range extraHeaders {
+		for _, value := range values {
+			cfg.Header.Add(header, value)
+		}
+	}
 
 	cfg.TlsConfig = st.tlsConfig
 	connection, err := websocketDialConfig(cfg)
@@ -414,7 +446,7 @@ func apiPath(modelTag names.ModelTag, path string) (string, error) {
 	if modelUUID == "" {
 		return path, nil
 	}
-	return "/model/" + modelUUID + path, nil
+	return modelRoot + modelUUID + path, nil
 }
 
 // tagToString returns the value of a tag's String method, or "" if the tag is nil.

api/base/caller.go

@@ -5,6 +5,7 @@ package base
 
 import (
 	"io"
+	"net/http"
 	"net/url"
 
 	"github.com/juju/httprequest"
@@ -39,6 +40,7 @@ type APICaller interface {
 	HTTPClient() (*httprequest.Client, error)
 
 	StreamConnector
+	ControllerStreamConnector
 }
 
 // StreamConnector is implemented by the client-facing State object.
@@ -53,6 +55,18 @@ type StreamConnector interface {
 	ConnectStream(path string, attrs url.Values) (Stream, error)
 }
 
+// ControllerStreamConnector is implemented by the client-facing State object.
+type ControllerStreamConnector interface {
+	// ConnectControllerStream connects to the given HTTP websocket
+	// endpoint path and returns the resulting connection. The given
+	// values are used as URL query values when making the initial
+	// HTTP request. Headers passed in will be added to the HTTP
+	// request.
+	//
+	// The path must be absolute and can't start with "/model".
+	ConnectControllerStream(path string, attrs url.Values, headers http.Header) (Stream, error)
+}
+
 // Stream represents a streaming connection to the API.
 type Stream interface {
 	io.ReadWriteCloser

api/base/testing/apicaller.go

@@ -4,6 +4,7 @@
 package testing
 
 import (
+	"net/http"
 	"net/url"
 
 	"github.com/juju/errors"
@@ -45,7 +46,11 @@ func (APICallerFunc) HTTPClient() (*httprequest.Client, error) {
 }
 
 func (APICallerFunc) ConnectStream(path string, attrs url.Values) (base.Stream, error) {
-	return nil, errors.New("stream connection unimplemented")
+	return nil, errors.NotImplementedf("stream connection")
+}
+
+func (APICallerFunc) ConnectControllerStream(path string, attrs url.Values, headers http.Header) (base.Stream, error) {
+	return nil, errors.NotImplementedf("controller stream connection")
 }
 
 // CheckArgs holds the possible arguments to CheckingAPICaller(). Any

api/client.go

@@ -12,10 +12,8 @@ import (
 	"os"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/juju/errors"
-	"github.com/juju/loggo"
 	"github.com/juju/version"
 	"golang.org/x/net/websocket"
 	"gopkg.in/juju/charm.v6-unstable"
@@ -24,6 +22,7 @@ import (
 	"gopkg.in/macaroon.v1"
 
 	"github.com/juju/juju/api/base"
+	"github.com/juju/juju/api/common"
 	"github.com/juju/juju/apiserver/params"
 	"github.com/juju/juju/constraints"
 	"github.com/juju/juju/downloader"
@@ -524,111 +523,8 @@ func (c websocketStream) WriteJSON(v interface{}) error {
 	return websocket.JSON.Send(c.Conn, v)
 }
 
-// TODO(ericsnow) Fold DebugLogParams into params.LogStreamConfig.
-
-// DebugLogParams holds parameters for WatchDebugLog that control the
-// filtering of the log messages. If the structure is zero initialized, the
-// entire log file is sent back starting from the end, and until the user
-// closes the connection.
-type DebugLogParams struct {
-	// IncludeEntity lists entity tags to include in the response. Tags may
-	// finish with a '*' to match a prefix e.g.: unit-mysql-*, machine-2. If
-	// none are set, then all lines are considered included.
-	IncludeEntity []string
-	// IncludeModule lists logging modules to include in the response. If none
-	// are set all modules are considered included.  If a module is specified,
-	// all the submodules also match.
-	IncludeModule []string
-	// ExcludeEntity lists entity tags to exclude from the response. As with
-	// IncludeEntity the values may finish with a '*'.
-	ExcludeEntity []string
-	// ExcludeModule lists logging modules to exclude from the resposne. If a
-	// module is specified, all the submodules are also excluded.
-	ExcludeModule []string
-	// Limit defines the maximum number of lines to return. Once this many
-	// have been sent, the socket is closed.  If zero, all filtered lines are
-	// sent down the connection until the client closes the connection.
-	Limit uint
-	// Backlog tells the server to try to go back this many lines before
-	// starting filtering. If backlog is zero and replay is false, then there
-	// may be an initial delay until the next matching log message is written.
-	Backlog uint
-	// Level specifies the minimum logging level to be sent back in the response.
-	Level loggo.Level
-	// Replay tells the server to start at the start of the log file rather
-	// than the end. If replay is true, backlog is ignored.
-	Replay bool
-	// NoTail tells the server to only return the logs it has now, and not
-	// to wait for new logs to arrive.
-	NoTail bool
-}
-
-func (args DebugLogParams) URLQuery() url.Values {
-	attrs := url.Values{
-		"includeEntity": args.IncludeEntity,
-		"includeModule": args.IncludeModule,
-		"excludeEntity": args.ExcludeEntity,
-		"excludeModule": args.ExcludeModule,
-	}
-	if args.Replay {
-		attrs.Set("replay", fmt.Sprint(args.Replay))
-	}
-	if args.NoTail {
-		attrs.Set("noTail", fmt.Sprint(args.NoTail))
-	}
-	if args.Limit > 0 {
-		attrs.Set("maxLines", fmt.Sprint(args.Limit))
-	}
-	if args.Backlog > 0 {
-		attrs.Set("backlog", fmt.Sprint(args.Backlog))
-	}
-	if args.Level != loggo.UNSPECIFIED {
-		attrs.Set("level", fmt.Sprint(args.Level))
-	}
-	return attrs
-}
-
-// LogMessage is a structured logging entry.
-type LogMessage struct {
-	Entity    string
-	Timestamp time.Time
-	Severity  string
-	Module    string
-	Location  string
-	Message   string
-}
-
 // WatchDebugLog returns a channel of structured Log Messages. Only log entries
 // that match the filtering specified in the DebugLogParams are returned.
-func (c *Client) WatchDebugLog(args DebugLogParams) (<-chan LogMessage, error) {
-	// Prepare URL query attributes.
-	attrs := args.URLQuery()
-
-	connection, err := c.st.ConnectStream("/log", attrs)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-
-	messages := make(chan LogMessage)
-	go func() {
-		defer close(messages)
-
-		for {
-			var msg params.LogMessage
-			err := connection.ReadJSON(&msg)
-			if err != nil {
-				return
-			}
-			messages <- LogMessage{
-				Entity:    msg.Entity,
-				Timestamp: msg.Timestamp,
-				Severity:  msg.Severity,
-				Module:    msg.Module,
-				Location:  msg.Location,
-				Message:   msg.Message,
-			}
-		}
-	}()
-
-	return messages, nil
+func (c *Client) WatchDebugLog(args common.DebugLogParams) (<-chan common.LogMessage, error) {
+	return common.StreamDebugLog(c.st, args)
 }

api/client_test.go

@@ -13,6 +13,7 @@ import (
 	"net/url"
 	"path"
 	"strings"
+	"time"
 
 	"github.com/juju/errors"
 	"github.com/juju/httprequest"
@@ -26,6 +27,7 @@ import (
 
 	"github.com/juju/juju/api"
 	"github.com/juju/juju/api/base"
+	"github.com/juju/juju/api/common"
 	"github.com/juju/juju/apiserver/params"
 	jujunames "github.com/juju/juju/juju/names"
 	jujutesting "github.com/juju/juju/juju/testing"
@@ -356,7 +358,7 @@ func (s *clientSuite) TestWatchDebugLogConnected(c *gc.C) {
 	// Use the no tail option so we don't try to start a tailing cursor
 	// on the oplog when there is no oplog configured in mongo as the tests
 	// don't set up mongo in replicaset mode.
-	messages, err := client.WatchDebugLog(api.DebugLogParams{NoTail: true})
+	messages, err := client.WatchDebugLog(common.DebugLogParams{NoTail: true})
 	c.Assert(err, jc.ErrorIsNil)
 	c.Assert(messages, gc.NotNil)
 }
@@ -404,11 +406,37 @@ func (s *clientSuite) TestConnectStreamErrorReadError(c *gc.C) {
 	c.Assert(reader, gc.IsNil)
 }
 
+func (s *clientSuite) TestConnectControllerStreamRejectsRelativePaths(c *gc.C) {
+	reader, err := s.APIState.ConnectControllerStream("foo", nil, nil)
+	c.Assert(err, gc.ErrorMatches, `path "foo" is not absolute`)
+	c.Assert(reader, gc.IsNil)
+}
+
+func (s *clientSuite) TestConnectControllerStreamRejectsModelPaths(c *gc.C) {
+	reader, err := s.APIState.ConnectControllerStream("/model/foo", nil, nil)
+	c.Assert(err, gc.ErrorMatches, `path "/model/foo" is model-specific`)
+	c.Assert(reader, gc.IsNil)
+}
+
+func (s *clientSuite) TestConnectControllerStreamAppliesHeaders(c *gc.C) {
+	catcher := urlCatcher{}
+	headers := http.Header{}
+	headers.Add("thomas", "cromwell")
+	headers.Add("anne", "boleyn")
+	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
+
+	_, err := s.APIState.ConnectControllerStream("/something", nil, headers)
+	c.Assert(err, jc.ErrorIsNil)
+
+	c.Assert(catcher.headers.Get("thomas"), gc.Equals, "cromwell")
+	c.Assert(catcher.headers.Get("anne"), gc.Equals, "boleyn")
+}
+
 func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
 	catcher := urlCatcher{}
 	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
 
-	params := api.DebugLogParams{
+	params := common.DebugLogParams{
 		IncludeEntity: []string{"a", "b"},
 		IncludeModule: []string{"c", "d"},
 		ExcludeEntity: []string{"e", "f"},
@@ -418,6 +446,7 @@ func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
 		Level:         loggo.ERROR,
 		Replay:        true,
 		NoTail:        true,
+		StartTime:     time.Date(2016, 11, 30, 11, 48, 0, 100, time.UTC),
 	}
 
 	client := s.APIState.Client()
@@ -436,6 +465,7 @@ func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
 		"level":         {"ERROR"},
 		"replay":        {"true"},
 		"noTail":        {"true"},
+		"startTime":     {"2016-11-30T11:48:00.0000001Z"},
 	})
 }
 
@@ -530,10 +560,12 @@ func (r *badReader) Read(p []byte) (n int, err error) {
 
 type urlCatcher struct {
 	location *url.URL
+	headers  http.Header
 }
 
 func (u *urlCatcher) recordLocation(config *websocket.Config) (base.Stream, error) {
 	u.location = config.Location
+	u.headers = config.Header
 	pr, pw := io.Pipe()
 	go func() {
 		fmt.Fprintf(pw, "null\n")