Method PushStream

pkg/ingester/ingester.go:1687–1737

(srv client.Ingester_PushStreamServer)

}

func (i *Ingester) PushStream(srv client.Ingester_PushStreamServer) error {
	ctx := srv.Context()
	for {
		select {
		case <-ctx.Done():
			level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "PushStream closed")
			return ctx.Err()
		default:
		}

		req, err := srv.Recv()

		if err == io.EOF {
			return nil
		}

		if err != nil {
			return err
		}

		if contextOrgID, extractErr := users.TenantID(ctx); extractErr == nil {
			if !isDistributorWorkerOrgID(contextOrgID) && contextOrgID != req.TenantID {
				req.Free()
				return status.Errorf(codes.PermissionDenied,
					"tenant ID mismatch: stream authenticated as %q but request specifies %q",
					contextOrgID, req.TenantID)
			}
		}

		pushCtx := user.InjectOrgID(ctx, req.TenantID)
		resp, err := i.Push(pushCtx, req.Request)
		if resp == nil {
			resp = &cortexpb.WriteResponse{}
		}
		resp.Code = http.StatusOK
		if err != nil {
			httpResponse, isGRPCError := httpgrpc.HTTPResponseFromError(err)
			if !isGRPCError {
				err = httpgrpc.Errorf(http.StatusInternalServerError, "%s", err)
				httpResponse, _ = httpgrpc.HTTPResponseFromError(err)
			}
			resp.Code = httpResponse.Code
			resp.Message = string(httpResponse.Body)
		}
		err = srv.Send(resp)
		req.Free()
		if err != nil {
			level.Error(logutil.WithContext(ctx, i.logger)).Log("msg", "error sending from PushStream", "err", err)
		}
	}
}

// isDistributorWorkerOrgID reports whether orgID matches the synthetic worker-name pattern
// that the distributor injects as X-Scope-OrgID when opening a long-lived PushStream: