// LogStream streams the build logs to the server.
769210
func (c *client) LogStream(id int64) (StreamWriter, error) {
	rawurl := fmt.Sprintf(pathLogsAuth, c.base, id, c.token)
	uri, err := url.Parse(rawurl)
	if err != nil {
		return nil, err
	}
	if uri.Scheme == "https" {
		uri.Scheme = "wss"
	} else {
		uri.Scheme = "ws"
	}

	// TODO need TLS client configuration

	conn, _, err := websocket.DefaultDialer.Dial(uri.String(), nil)
	return conn, err
}
558870
func newLogsStreamer(c *Context) (empire.LogsStreamer, error) {
	switch c.String(FlagLogsStreamer) {
	case "kinesis":
		return newKinesisLogsStreamer(c)
	default:
		log.Println("Streaming logs are disabled")
		return nil, nil
	}
}
321390
func LogStreamer() http.Handler {
	logs := mux.NewRouter()
	logsHandler := func(w http.ResponseWriter, req *http.Request) {
		params := mux.Vars(req)
		route := new(router.Route)

		if params["value"] != "" {
			switch params["predicate"] {
			case "id":
				route.FilterID = params["value"]
				if len(route.ID) > 12 {
					route.FilterID = route.FilterID[:12]
				}
			case "name":
				route.FilterName = params["value"]
			}
		}

		if route.FilterID != "" && !router.Routes.RoutingFrom(route.FilterID) {
			http.NotFound(w, req)
			return
		}

		defer debug("http: logs streamer disconnected")
		logstream := make(chan *router.Message)
		defer close(logstream)

		var closer <-chan bool
		if req.Header.Get("Upgrade") == "websocket" {
			debug("http: logs streamer connected [websocket]")
			closerBi := make(chan bool)
			defer websocketStreamer(w, req, logstream, closerBi)
			closer = closerBi
		} else {
			debug("http: logs streamer connected [http]")
			defer httpStreamer(w, req, logstream, route.MultiContainer())
			closer = w.(http.CloseNotifier).CloseNotify()
		}
		route.OverrideCloser(closer)

		router.Routes.Route(route, logstream)
	}
	logs.HandleFunc("/logs/{predicate:[a-zA-Z]+}:{value}", logsHandler).Methods("GET")
	logs.HandleFunc("/logs", logsHandler).Methods("GET")
	return logs
}
432253
func (l *LogSender) ScanLogStream(appID, sourceType, sourceInstance string, reader io.Reader) {
	l.scanLogStream(appID, sourceType, sourceInstance, l.SendAppLog, reader)
}
183717
func MakeStreamLogger(ws WriteStringer) Sender {
	s := &streamLogger{
		fobj: ws,
		Base: NewBase(""),
	}

	fallback := log.New(os.Stdout, "", log.LstdFlags)
	_ = s.SetErrorHandler(ErrorHandlerFromLogger(fallback))

	s.reset = func() {
		fallback.SetPrefix(fmt.Sprintf("[%s] ", s.Name()))
	}

	return s
}
1146414
func WrapServerStream(stream grpc.ServerStream) *WrappedServerStream {
	if existing, ok := stream.(*WrappedServerStream); ok {
		return existing
	}
	return &WrappedServerStream{ServerStream: stream, WrappedContext: stream.Context()}
}
157528
func (c *Client) OpenLogTransferStream(modelUUID string) (base.Stream, error) {
	attrs := url.Values{}
	attrs.Set("jujuclientversion", jujuversion.Current.String())
	headers := http.Header{}
	headers.Set(params.MigrationModelHTTPHeader, modelUUID)
	caller := c.caller.RawAPICaller()
	stream, err := caller.ConnectControllerStream("/migrate/logtransfer", attrs, headers)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return stream, nil
}
389140
func (lb *LogBroker) PublishLogs(stream api.LogBroker_PublishLogsServer) (err error) {
	remote, err := ca.RemoteNode(stream.Context())
	if err != nil {
		return err
	}

	var currentSubscription *subscription
	defer func() {
		if currentSubscription != nil {
			lb.markDone(currentSubscription, remote.NodeID, err)
		}
	}()

	for {
		logMsg, err := stream.Recv()
		if err == io.EOF {
			return stream.SendAndClose(&api.PublishLogsResponse{})
		}
		if err != nil {
			return err
		}

		if logMsg.SubscriptionID == "" {
			return status.Errorf(codes.InvalidArgument, "missing subscription ID")
		}

		if currentSubscription == nil {
			currentSubscription = lb.getSubscription(logMsg.SubscriptionID)
			if currentSubscription == nil {
				return status.Errorf(codes.NotFound, "unknown subscription ID")
			}
		} else {
			if logMsg.SubscriptionID != currentSubscription.message.ID {
				return status.Errorf(codes.InvalidArgument, "different subscription IDs in the same session")
			}
		}

		// if we have a close message, close out the subscription
		if logMsg.Close {
			// Mark done and then set to nil so if we error after this point,
			// we don't try to close again in the defer
			lb.markDone(currentSubscription, remote.NodeID, err)
			currentSubscription = nil
			return nil
		}

		// Make sure logs are emitted using the right Node ID to avoid impersonation.
		for _, msg := range logMsg.Messages {
			if msg.Context.NodeID != remote.NodeID {
				return status.Errorf(codes.PermissionDenied, "invalid NodeID: expected=%s;received=%s", remote.NodeID, msg.Context.NodeID)
			}
		}

		lb.publish(logMsg)
	}
}
1043527
func logPath(dir, id, stream string) string {
	return filepath.Join(dir, fmt.Sprintf("%s-%s.log", id, stream))
}
900550
func (f *FileSystem) stream(conn io.ReadWriteCloser) {
	defer conn.Close()
	defer metrics.MeasureSince([]string{"nomad", "file_system", "stream"}, time.Now())

	// Decode the arguments
	var args cstructs.FsStreamRequest
	decoder := codec.NewDecoder(conn, structs.MsgpackHandle)
	encoder := codec.NewEncoder(conn, structs.MsgpackHandle)

	if err := decoder.Decode(&args); err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(500), encoder)
		return
	}

	// Check if we need to forward to a different region
	if r := args.RequestRegion(); r != f.srv.Region() {
		f.forwardRegionStreamingRpc(conn, encoder, &args, "FileSystem.Stream",
			args.AllocID, &args.QueryOptions)
		return
	}

	// Check node read permissions
	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
		f.handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
		return
	}

	// Verify the arguments.
	if args.AllocID == "" {
		f.handleStreamResultError(errors.New("missing AllocID"), helper.Int64ToPtr(400), encoder)
		return
	}

	// Retrieve the allocation
	snap, err := f.srv.State().Snapshot()
	if err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	}

	alloc, err := snap.AllocByID(nil, args.AllocID)
	if err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	}
	if alloc == nil {
		f.handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
		return
	}
	nodeID := alloc.NodeID

	// Make sure Node is valid and new enough to support RPC
	node, err := snap.NodeByID(nil, nodeID)
	if err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(500), encoder)
		return
	}

	if node == nil {
		err := fmt.Errorf("Unknown node %q", nodeID)
		f.handleStreamResultError(err, helper.Int64ToPtr(400), encoder)
		return
	}

	if err := nodeSupportsRpc(node); err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(400), encoder)
		return
	}

	// Get the connection to the client either by forwarding to another server
	// or creating a direct stream
	var clientConn net.Conn
	state, ok := f.srv.getNodeConn(nodeID)
	if !ok {
		// Determine the Server that has a connection to the node.
		srv, err := f.srv.serverWithNodeConn(nodeID, f.srv.Region())
		if err != nil {
			var code *int64
			if structs.IsErrNoNodeConn(err) {
				code = helper.Int64ToPtr(404)
			}
			f.handleStreamResultError(err, code, encoder)
			return
		}

		// Get a connection to the server
		conn, err := f.srv.streamingRpc(srv, "FileSystem.Stream")
		if err != nil {
			f.handleStreamResultError(err, nil, encoder)
			return
		}

		clientConn = conn
	} else {
		stream, err := NodeStreamingRpc(state.Session, "FileSystem.Stream")
		if err != nil {
			f.handleStreamResultError(err, nil, encoder)
			return
		}
		clientConn = stream
	}
	defer clientConn.Close()

	// Send the request.
	outEncoder := codec.NewEncoder(clientConn, structs.MsgpackHandle)
	if err := outEncoder.Encode(args); err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	}

	structs.Bridge(conn, clientConn)
	return
}
973074
func (c *Connect) NewStream(flags StreamFlags) (*Stream, error) {
	var err C.virError
	virStream := C.virStreamNewWrapper(c.ptr, C.uint(flags), &err)
	if virStream == nil {
		return nil, makeError(&err)
	}

	return &Stream{
		ptr: virStream,
	}, nil
}
694194
func (sf StreamerFunc) Stream(samples [][2]float64) (n int, ok bool) {
	return sf(samples)
}
304377
func (api *API) LogWriter() (LogWriter, error) {
	attrs := make(url.Values)
	attrs.Set("jujuclientversion", version.Current.String())
	// Version 1 does ping/pong handling.
	attrs.Set("version", "1")
	conn, err := api.connector.ConnectStream("/logsink", attrs)
	if err != nil {
		return nil, errors.Annotatef(err, "cannot connect to /logsink")
	}
	logWriter := writer{conn}
	go logWriter.readLoop()
	return logWriter, nil
}
873170
func (c *Collector) processLogStream(ctx context.Context, h *bundleEntryHandler) error {
	streamTypeField := streamType(h.be.Desc)
	startTime := clock.Now(ctx)
	defer func() {
		duration := clock.Now(ctx).Sub(startTime)

		// We track processing time in milliseconds.
		tsBundleEntryProcessingTime.Add(ctx, duration.Seconds()*1000, streamTypeField)
	}()

	// If this bundle has neither log entries nor a terminal index, it is junk and
	// must be discarded.
	//
	// This is more important than a basic optimization, as it enforces that no
	// zero-entry log streams can be ingested. Either some entries exist, or there
	// is a promise of a terminal entry.
	if len(h.be.Logs) == 0 && !h.be.Terminal {
		log.Warningf(ctx, "Bundle entry is non-terminal and contains no logs; discarding.")
		return nil
	}

	secret := types.PrefixSecret(h.b.Secret)
	if err := secret.Validate(); err != nil {
		log.Fields{
			log.ErrorKey:   err,
			"secretLength": len(secret),
		}.Errorf(ctx, "Failed to validate prefix secret.")
		return errors.New("invalid prefix secret")
	}

	// If the descriptor has a Prefix, it must match the bundle's Prefix.
	if p := h.be.Desc.Prefix; p != "" {
		if p != h.b.Prefix {
			log.Fields{
				"bundlePrefix":      h.b.Prefix,
				"bundleEntryPrefix": p,
			}.Errorf(ctx, "Bundle prefix does not match entry prefix.")
			return errors.New("mismatched bundle and entry prefixes")
		}
	} else {
		// Fill in the bundle's Prefix.
		h.be.Desc.Prefix = h.b.Prefix
	}

	if err := h.be.Desc.Validate(true); err != nil {
		log.WithError(err).Errorf(ctx, "Invalid log stream descriptor.")
		return err
	}
	descBytes, err := proto.Marshal(h.be.Desc)
	if err != nil {
		log.WithError(err).Errorf(ctx, "Failed to marshal descriptor.")
		return err
	}

	h.path = types.StreamName(h.be.Desc.Prefix).Join(types.StreamName(h.be.Desc.Name))
	ctx = log.SetFields(ctx, log.Fields{
		"project": h.project,
		"path":    h.path,
	})

	// Confirm that the log entries are valid and contiguous. Serialize the log
	// entries for ingest as we validate them.
	var logData [][]byte
	var blockIndex uint64
	if logs := h.be.Logs; len(logs) > 0 {
		logData = make([][]byte, len(logs))
		blockIndex = logs[0].StreamIndex

		for i, le := range logs {
			// Validate this log entry.
			if err := le.Validate(h.be.Desc); err != nil {
				log.Fields{
					log.ErrorKey: err,
					"index":      le.StreamIndex,
				}.Warningf(ctx, "Discarding invalid log entry.")
				return errors.New("invalid log entry")
			}

			// Validate that this entry is contiguous.
			if le.StreamIndex != blockIndex+uint64(i) {
				log.Fields{
					"index":    i,
					"expected": (blockIndex + uint64(i)),
					"actual":   le.StreamIndex,
				}.Errorf(ctx, "Non-contiguous log entry block in stream.")
				return errors.New("non-contiguous log entry block")
			}

			var err error
			logData[i], err = proto.Marshal(le)
			if err != nil {
				log.Fields{
					log.ErrorKey: err,
					"index":      le.StreamIndex,
				}.Errorf(ctx, "Failed to marshal log entry.")
				return errors.New("failed to marshal log entries")
			}
		}
	}

	// Fetch our cached/remote state. This will replace our state object with the
	// fetched state, so any future calls will need to re-set the Secret value.
	// TODO: Use timeout?
	registerReq := coordinator.LogStreamState{
		Project:       h.project,
		Path:          h.path,
		Secret:        secret,
		ProtoVersion:  h.md.ProtoVersion,
		TerminalIndex: -1,
	}
	if h.be.Terminal {
		registerReq.TerminalIndex = types.MessageIndex(h.be.TerminalIndex)
	}
	state, err := c.Coordinator.RegisterStream(ctx, &registerReq, descBytes)
	if err != nil {
		log.WithError(err).Errorf(ctx, "Failed to get/register current stream state.")
		return err
	}

	// Does the log stream's secret match the expected secret?
	//
	// Note that this check does NOT use the "subtle" package to do time-constant
	// byte comparison, and may leak inforamtion about the secret. This is OK,
	// since users cannot interact with this service directly; however, if this
	// code is ever used elsewhere, this should be a consideration.
	if !bytes.Equal([]byte(secret), []byte(state.Secret)) {
		log.Errorf(log.SetFields(ctx, log.Fields{
			"secret":         secret,
			"expectedSecret": state.Secret,
		}), "Log entry has incorrect secret.")
		return nil
	}

	if state.Archived {
		log.Infof(ctx, "Skipping message bundle for archived stream.")
		return nil
	}
	if state.Purged {
		log.Infof(ctx, "Skipping message bundle for purged stream.")
		return nil
	}

	// Update our terminal index if we have one.
	//
	// Note that even if our cached value is marked terminal, we could have failed
	// to push the terminal index to the Coordinator, so we will not refrain from
	// pushing every terminal index encountered regardless of cache state.
	if h.be.Terminal {
		tidx := types.MessageIndex(h.be.TerminalIndex)
		log.Fields{
			"value": tidx,
		}.Debugf(ctx, "Bundle includes a terminal index.")

		if state.TerminalIndex < 0 {
			state.TerminalIndex = tidx
		} else if state.TerminalIndex != tidx {
			log.Fields{
				"cachedIndex": state.TerminalIndex,
				"bundleIndex": tidx,
			}.Warningf(ctx, "Cached terminal index disagrees with state.")
		}
	}

	// Perform stream processing operations. We can do these operations in
	// parallel.
	return parallel.FanOutIn(func(taskC chan<- func() error) {
		// Store log data, if any was provided. It has already been validated.
		if len(logData) > 0 {
			taskC <- func() error {
				// Post the log to storage.
				err = c.Storage.Put(ctx, storage.PutRequest{
					Project: h.project,
					Path:    h.path,
					Index:   types.MessageIndex(blockIndex),
					Values:  logData,
				})

				// If the log entry already exists, consider the "put" successful.
				// Storage will return a transient error if one occurred.
				if err != nil && err != storage.ErrExists {
					log.Fields{
						log.ErrorKey: err,
						"blockIndex": blockIndex,
					}.Errorf(ctx, "Failed to load log entry into Storage.")
					return err
				}

				tsLogs.Add(ctx, int64(len(logData)))
				return nil
			}
		}

		// If our bundle entry is terminal, we have an additional task of reporting
		// this to the Coordinator.
		if h.be.Terminal {
			taskC <- func() error {
				// Sentinel task: Update the terminal bundle state.
				treq := coordinator.TerminateRequest{
					Project:       state.Project,
					Path:          state.Path,
					ID:            state.ID,
					Secret:        state.Secret,
					TerminalIndex: types.MessageIndex(h.be.TerminalIndex),
				}

				log.Fields{
					"terminalIndex": state.TerminalIndex,
				}.Infof(ctx, "Received terminal log; updating Coordinator state.")
				if err := c.Coordinator.TerminateStream(ctx, &treq); err != nil {
					log.WithError(err).Errorf(ctx, "Failed to set stream terminal index.")
					return err
				}
				return nil
			}
		}
	})
}
119737
func (m *Middleware) HandleStream(serverStream *transport.ServerStream, h transport.StreamHandler) error {
	call := m.graph.begin(serverStream.Context(), transport.Streaming, _directionInbound, serverStream.Request().Meta.ToRequest())
	err := h.HandleStream(serverStream)
	// TODO(pedge): wrap the *transport.ServerStream?
	call.End(err)
	return err
}
264844
func (s *gRPCBrokerServer) StartStream(stream plugin.GRPCBroker_StartStreamServer) error {
	doneCh := stream.Context().Done()
	defer s.Close()

	// Proccess send stream
	go func() {
		for {
			select {
			case <-doneCh:
				return
			case <-s.quit:
				return
			case se := <-s.send:
				err := stream.Send(se.i)
				se.ch <- err
			}
		}
	}()

	// Process receive stream
	for {
		i, err := stream.Recv()
		if err != nil {
			return err
		}
		select {
		case <-doneCh:
			return nil
		case <-s.quit:
			return nil
		case s.recv <- i:
		}
	}

	return nil
}
790300
func (a *Adapter) Stream(logstream chan *router.Message) {
	for message := range logstream {
		buf := new(bytes.Buffer)
		err := a.tmpl.Execute(buf, message)
		if err != nil {
			log.Println("raw:", err)
			return
		}
		//log.Println("debug:", buf.String())
		_, err = a.conn.Write(buf.Bytes())
		if err != nil {
			log.Println("raw:", err)
			if reflect.TypeOf(a.conn).String() != "*net.UDPConn" {
				return
			}
		}
	}
}