feat: structured log gathering from exporters (JEP-0013) - #930
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThis PR adds telemetry protobuf contracts, controller endpoint discovery, authenticated Go telemetry ingestion, and Python exporter log forwarding. It also consolidates controller configuration and adds a standalone telemetry service command with graceful shutdown. ChangesTelemetry collection
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (1)
python/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py (1)
1-267: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd an end-to-end test alongside the mocked unit tests.
All tests here mock
TelemetryServiceStub.PushLogs. Per repo guidelines for*_test.py, end-to-end tests that start a real server and client should be prioritized, with mocks reserved for cases where e2e is impractical. SinceTelemetryServiceServicercan be run in-process with a local gRPC server, an e2e test validating a fullTelemetryLogHandler→ real stub → real servicer round trip would strengthen coverage of the wire contract (serialization, field mapping) beyond what mocks can verify.As per coding guidelines, "Provide comprehensive package test coverage, prioritizing end-to-end tests that start a server and client; use mocks when system tools, services, or platform compatibility make end-to-end testing impractical."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py` around lines 1 - 267, Add an end-to-end test in TestFlush that starts an in-process local gRPC server with TelemetryServiceServicer, creates a real TelemetryServiceStub and TelemetryLogHandler, emits a record, and flushes it. Assert the servicer receives the serialized entry with the expected message, severity, and mapped fields, while keeping mocked tests for failure and queue behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controller/cmd/telemetry/main.go`:
- Around line 78-86: Update the signal-handling branch in main’s select to wait
for the telemetry service shutdown after calling cancel(). Receive from errCh
before returning so TelemetryService.Start() can execute GracefulStop() and
drain active RPCs; preserve the existing error logging and exit behavior for
service failures.
- Around line 66-72: Update the telemetry service startup around
TelemetryService and svc.Start to configure TLS for the ingress listener,
require and validate client certificates, and authorize authenticated clients
before accepting batches. Ensure discovery publishes the matching client
credentials and TLS settings so exporters connect securely without insecure
mode; preserve the existing service lifecycle and error propagation.
In `@controller/internal/service/telemetry_service.go`:
- Around line 85-111: Update TelemetryService.Start to configure grpc.NewServer
with TLS credentials derived from TelemetryConfig.Certificate and its
corresponding private key before registering the service, ensuring the server
uses the same certificate advertised by GetServiceEndpoints. Preserve the
existing lifecycle and graceful-shutdown behavior, and propagate
certificate-loading or server-setup errors.
- Around line 43-82: Bound unauthenticated batches in TelemetryService.PushLogs
by enforcing a server-side maximum number of entries before iterating and
logging them. Process only entries within the limit, report the remainder
through the response’s Dropped field, and preserve Accepted for processed
entries; use an established configuration or constant for the limit if
available.
In `@protocol/proto/jumpstarter/v1/telemetry.proto`:
- Around line 30-40: Update TelemetryService.PushLogs to preserve each log
entry’s Timestamp and Severity in the sink output. Use entry.Timestamp as the
emitted event timestamp rather than ingestion time, and retain entry.Severity as
a structured field so warning and debug entries remain distinguishable.
In
`@python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/client/v1/client_pb2.py`:
- Around line 33-34: Regenerate client_pb2.py and the related protobuf classes
using one consistent protoc/grpcio-tools configuration, ensuring its
kubernetes_pb2 and common_pb2 imports match the relative ..v1 imports used by
the other generated pb2 files. Do not hand-edit the generated output; update the
generation process or invocation if needed, then regenerate the affected files
consistently.
In `@python/packages/jumpstarter/jumpstarter/exporter/exporter.py`:
- Around line 1181-1188: Update the shutdown flow around
_telemetry_handler.close_async() to drain all queued telemetry batches before
closing _telemetry_channel. Repeat flushing until the handler’s queue is empty,
then proceed with the existing handler removal and channel closure.
- Around line 408-434: Update the telemetry setup around TelemetryLogHandler so
it receives ep.min_severity when constructed, and enforce that threshold when
processing exporter records. Preserve the existing channel selection and handler
registration, ensuring records below the controller-advertised minimum severity
are filtered out.
- Around line 393-397: Bound both best-effort telemetry RPCs with _RPC_TIMEOUT:
pass the deadline when awaiting GetServiceEndpoints in the exporter registration
flow, and apply the same deadline to PushLogs in TelemetryLogHandler._flush().
Preserve existing error handling while ensuring registration, periodic flushing,
and shutdown cannot wait indefinitely.
---
Nitpick comments:
In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py`:
- Around line 1-267: Add an end-to-end test in TestFlush that starts an
in-process local gRPC server with TelemetryServiceServicer, creates a real
TelemetryServiceStub and TelemetryLogHandler, emits a record, and flushes it.
Assert the servicer receives the serialized entry with the expected message,
severity, and mapped fields, while keeping mocked tests for failure and queue
behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 7210c4a5-f9a9-4494-a887-2484a984d27d
⛔ Files ignored due to path filters (4)
controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/telemetry.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/telemetry_grpc.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (23)
controller/cmd/main.gocontroller/cmd/telemetry/main.gocontroller/deploy/operator/config/crd/bases/virtualtarget.jumpstarter.dev_exportersets.yamlcontroller/internal/config/config.gocontroller/internal/config/types.gocontroller/internal/service/controller_service.gocontroller/internal/service/telemetry_service.gocontroller/internal/service/telemetry_service_test.goprotocol/proto/jumpstarter/v1/jumpstarter.protoprotocol/proto/jumpstarter/v1/telemetry.protopython/packages/jumpstarter-protocol/jumpstarter_protocol/__init__.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/client/v1/client_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pyipython/packages/jumpstarter/jumpstarter/exporter/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
python/packages/jumpstarter/jumpstarter/exporter/exporter.py (1)
404-414: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winKeep all optional telemetry setup failures non-fatal.
The handler catches only
grpc.aio.AioRpcErrorfrom endpoint discovery.channel_factory()and the subsequent telemetry channel/handler setup can raise other exceptions; since_register_with_controller()awaits this after registration, those failures can abort exporter startup despite telemetry being optional. Wrap the complete setup in best-effort handling and clean up any partially created telemetry resources.Also applies to: 429-447
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/packages/jumpstarter/jumpstarter/exporter/exporter.py` around lines 404 - 414, Update _register_with_controller around endpoint discovery and telemetry setup so every optional telemetry failure, including channel_factory and channel/handler construction errors, is caught and cannot abort exporter startup. Wrap the complete setup in best-effort exception handling and clean up any partially created telemetry resources before returning on failure, while preserving successful registration behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controller/internal/service/telemetry_service.go`:
- Around line 120-122: Update the context-cancellation branch around
srv.GracefulStop() to normalize grpc.ErrServerStopped from errCh as a successful
shutdown when ctx is canceled. Preserve and return other Serve errors unchanged.
---
Outside diff comments:
In `@python/packages/jumpstarter/jumpstarter/exporter/exporter.py`:
- Around line 404-414: Update _register_with_controller around endpoint
discovery and telemetry setup so every optional telemetry failure, including
channel_factory and channel/handler construction errors, is caught and cannot
abort exporter startup. Wrap the complete setup in best-effort exception
handling and clean up any partially created telemetry resources before returning
on failure, while preserving successful registration behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1fea4c4f-3b1c-4e0a-818d-db584454eee2
📒 Files selected for processing (6)
controller/cmd/telemetry/main.gocontroller/internal/config/config.gocontroller/internal/config/types.gocontroller/internal/service/telemetry_service.gopython/packages/jumpstarter/jumpstarter/exporter/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry.py
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
python/packages/jumpstarter/jumpstarter/exporter/exporter.py (1)
404-414: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winKeep optional telemetry setup non-fatal.
Only
AioRpcErroris handled. An ordinary exception from_controller_stub()can escape after registration and abort the exporter, despite telemetry being best-effort. CatchExceptionhere as well, while allowing cancellation (BaseException) through.Proposed fix
except grpc.aio.AioRpcError as e: logger.debug("GetServiceEndpoints unavailable: %s", e.code()) return + except Exception: + logger.debug("GetServiceEndpoints setup failed", exc_info=True) + return🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/packages/jumpstarter/jumpstarter/exporter/exporter.py` around lines 404 - 414, Update the GetServiceEndpoints setup around _controller_stub() to catch ordinary Exception failures in addition to grpc.aio.AioRpcError, log the failure at debug level, and return without aborting exporter registration. Keep cancellation and other BaseException subclasses propagating, and preserve the existing successful response flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/packages/jumpstarter/jumpstarter/exporter/exporter_telemetry_test.py`:
- Around line 33-42: Update the telemetry setup tests using make_bare_exporter
to track the root logger handler created by each test, then add teardown cleanup
that removes and closes that handler after every test. Ensure cleanup runs even
when a test fails and does not affect unrelated root logger handlers.
In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry.py`:
- Around line 134-137: Update close_async so shutdown does not retry every
queued batch after telemetry becomes unavailable. Track whether _flush()
successfully pushed the batch and stop draining immediately after the first
failure, or enforce a single overall shutdown deadline while preserving the
existing flush behavior.
---
Outside diff comments:
In `@python/packages/jumpstarter/jumpstarter/exporter/exporter.py`:
- Around line 404-414: Update the GetServiceEndpoints setup around
_controller_stub() to catch ordinary Exception failures in addition to
grpc.aio.AioRpcError, log the failure at debug level, and return without
aborting exporter registration. Keep cancellation and other BaseException
subclasses propagating, and preserve the existing successful response flow.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: f51c6ad4-021b-4305-b619-e4e28a24c309
📒 Files selected for processing (6)
controller/cmd/telemetry/main.gocontroller/internal/service/telemetry_service.gopython/packages/jumpstarter/jumpstarter/exporter/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/exporter_telemetry_test.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
| def make_bare_exporter(): | ||
| """Minimal Exporter instance for testing _setup_telemetry in isolation.""" | ||
| exp = Exporter.__new__(Exporter) | ||
| exp._telemetry_handler = None | ||
| exp._telemetry_channel = None | ||
|
|
||
| tls = MagicMock() | ||
| tls.insecure = False | ||
| exp.tls = tls | ||
| return exp |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Clean up root handlers after each setup test.
Successful setup tests register a handler on the global root logger but never remove it. Handlers accumulate across the module and can capture later tests’ logs. Add teardown that removes and closes the handler created by each test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/packages/jumpstarter/jumpstarter/exporter/exporter_telemetry_test.py`
around lines 33 - 42, Update the telemetry setup tests using make_bare_exporter
to track the root logger handler created by each test, then add teardown cleanup
that removes and closes that handler after every test. Ensure cleanup runs even
when a test fails and does not affect unrelated root logger handlers.
| async def close_async(self) -> None: | ||
| """Flush all remaining entries and close the handler.""" | ||
| while self._queue: | ||
| await self._flush() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not multiply shutdown time by queued batches.
_flush() drops a failed batch but hides the failure. With telemetry unavailable, this loop waits up to 10 seconds for each queued batch before exit. Stop draining after the first failed push, or apply one total shutdown deadline.
Proposed fix
- async def _flush(self) -> None:
+ async def _flush(self) -> bool:
...
try:
await self._stub.PushLogs(
telemetry_pb2.PushLogsRequest(entries=batch),
timeout=_PUSH_TIMEOUT,
)
except Exception as exc:
print(f"[telemetry] PushLogs failed, {len(batch)} entries dropped: {exc}", file=sys.stderr)
+ return False
+ return True
async def close_async(self) -> None:
"""Flush all remaining entries and close the handler."""
while self._queue:
- await self._flush()
+ if not await self._flush():
+ break
self.close()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def close_async(self) -> None: | |
| """Flush all remaining entries and close the handler.""" | |
| while self._queue: | |
| await self._flush() | |
| async def close_async(self) -> None: | |
| """Flush all remaining entries and close the handler.""" | |
| while self._queue: | |
| if not await self._flush(): | |
| break | |
| self.close() |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry.py` around lines
134 - 137, Update close_async so shutdown does not retry every queued batch
after telemetry becomes unavailable. Track whether _flush() successfully pushed
the batch and stop draining immediately after the first failure, or enforce a
single overall shutdown deadline while preserving the existing flush behavior.
RoddieKieley
left a comment
RoddieKieley
left a comment
There was a problem hiding this comment.
Mostly nit-pick type comments on minor details for the moment. May have more detailed comments after building the branch and testing it in-cluster.
| var configmap corev1.ConfigMap | ||
| if err := client.Get(ctx, key, &configmap); err != nil { | ||
| return nil, "", nil, nil, nil, nil, nil, err | ||
| return nil, "", nil, nil, nil, nil, nil, nil, err |
There was a problem hiding this comment.
I've noticed that the LLM's don't mind not being dry, an in this case there's quite a few of this particular return value repeated. However prior to this code change it was only 1 nil different so for the moment it's probably fine without the refactor. This is probably one of those areas where adopting some clean code type rules to guide the coding agent up front could help.
There was a problem hiding this comment.
Yes, a few PRs ago we had the same conversation with HiddenLabels.
It could be worth to return all this as a config structure, and then just pass the structure to the controller later in https://github.com/jumpstarter-dev/jumpstarter/pull/930/changes#diff-032eafab48326b32dad354dd24c3da532106bde78b3589793bd8fdc47be0bb65R283 as a single thing, we could also return "nil" config :) single nil
There was a problem hiding this comment.
Done, added LoadConfiguration now returns (*LoadedConfig, error), a single struct that carries all the previously separate return values. All error paths return a single nil, err. The call site in main.go was updated accordingly.
| Certificate string `json:"certificate,omitempty" yaml:"certificate,omitempty"` | ||
|
|
||
| // MinSeverity is the minimum log severity to forward to the telemetry service. | ||
| // Accepted values: debug, info, warning, error. Defaults to "info" when empty. |
There was a problem hiding this comment.
Is this correct? Is there also a critical that is mentioned elsewhere in the code?
"
"""Normalise Python logging level names to JEP-0013 severity strings."""
return {
"DEBUG": "debug",
"INFO": "info",
"WARNING": "warning",
"ERROR": "error",
"CRITICAL": "critical",
"
There was a problem hiding this comment.
def _severity_to_level(severity: str) -> int:
|
@bkhizgiy I ran up your exporter-logs branch on top of my local setup here and confirm local exporter logging output: matches the log output as viewed via LogQL inside the local openshift sno cluster: I did have to manually configure and expose a Service to connect to the jumpstarter-telemetry pods endpoint but I think that's expected at this point of the phased implementation of JEP-0013. |
| var configmap corev1.ConfigMap | ||
| if err := client.Get(ctx, key, &configmap); err != nil { | ||
| return nil, "", nil, nil, nil, nil, nil, err | ||
| return nil, "", nil, nil, nil, nil, nil, nil, err |
There was a problem hiding this comment.
Yes, a few PRs ago we had the same conversation with HiddenLabels.
It could be worth to return all this as a config structure, and then just pass the structure to the controller later in https://github.com/jumpstarter-dev/jumpstarter/pull/930/changes#diff-032eafab48326b32dad354dd24c3da532106bde78b3589793bd8fdc47be0bb65R283 as a single thing, we could also return "nil" config :) single nil
| Certificate string `json:"certificate,omitempty" yaml:"certificate,omitempty"` | ||
|
|
||
| // MinSeverity is the minimum log severity to forward to the telemetry service. | ||
| // Accepted values: debug, info, warning, error. Defaults to "info" when empty. |
There was a problem hiding this comment.
def _severity_to_level(severity: str) -> int:
| Name is the key used in the ExporterConfig export map. | ||
| If omitted, derived from the Type (last segment after the last dot). | ||
| description: Name is the key used in the ExporterConfig | ||
| export map. |
There was a problem hiding this comment.
haha yep, good eye was removed.
| // controller-runtime logger (structured JSON to stdout). | ||
| // Future phase: forward to Loki push API. | ||
| func (s *TelemetryService) PushLogs(ctx context.Context, req *pb.PushLogsRequest) (*pb.PushLogsResponse, error) { | ||
| logger := ctrl.Log.WithName("telemetry") |
There was a problem hiding this comment.
we need to authenticate the exporter from the call.
There was a problem hiding this comment.
Done, PushLogs now validates the bearer token and rejects any token whose subject is not in exporter:namespace:name:uid form (client tokens and other validly signed tokens are rejected with PermissionDenied). Entries whose exporter or namespace fields don't match the token's identity are counted as dropped rather than failing the whole batch.
| LeasePolicy LeasePolicy `json:"leasePolicy,omitempty" yaml:"leasePolicy,omitempty"` | ||
| HiddenLabels HiddenLabels `json:"hiddenLabels,omitempty" yaml:"hiddenLabels,omitempty"` | ||
| Telemetry *Telemetry `json:"telemetry,omitempty" yaml:"telemetry,omitempty"` | ||
| } |
There was a problem hiding this comment.
The proposed JEP configuration is quite different: https://jumpstarter.dev/main/contributing/jeps/JEP-0013-observability-telemetry-logs.html#operator-configuration
At least the logging related should match.
Also, the JEP was missing endpoint config, but we should re-use the structs from the router/controller/etc. And make it auto-calculate the endpoint name as we do with the others.
enabled/disabled must be controlled by the telemetry.enabled flag.
There was a problem hiding this comment.
good point, the telemetry.logging.filter structure now matches the JEP.
Endpoint auto-derivation was implemented when the endpoint is empty and enabled: true, the controller derives jumpstarter-telemetry.<namespace>:9093 (the same pattern as the router and controller services). The Certificate field is reserved for Phase 2 TLS support and is now clearly documented as such.
And regarding enabled/disabled, GetServiceEndpoints only advertises the telemetry endpoint when telemetry.enabled: true.
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
python/packages/jumpstarter/jumpstarter/exporter/telemetry.py (1)
154-158: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winShutdown can block for
_PUSH_TIMEOUTper queued batch.
_flushreturnsNoneon both success and failure, soclose_asynccannot tell them apart. If the telemetry service is unreachable, each iteration waits up to_PUSH_TIMEOUT(10 s) before dropping its batch. With a full queue of 10,000 entries and_BATCH_SIZEof 50, that is 200 iterations.The loop also has no termination guarantee while the handler stays attached to the root logger, because
emitcan refillself._queueduring the drain.Make
_flushreport success and stop the drain after the first failure, or apply one total deadline.🛠️ Proposed fix
- async def _flush(self) -> None: + async def _flush(self) -> bool: """Drain up to _BATCH_SIZE entries from the queue and push to telemetry.""" batch: list[telemetry_pb2.LogEntry] = [] while self._queue and len(batch) < _BATCH_SIZE: batch.append(self._queue.popleft()) if not batch: - return + return True metadata = [("authorization", f"Bearer {self._token}")] if self._token else [] try: await self._stub.PushLogs( telemetry_pb2.PushLogsRequest(entries=batch), timeout=_PUSH_TIMEOUT, metadata=metadata, ) except Exception as exc: # Avoid recursive logging: write directly to stderr. print(f"[telemetry] PushLogs failed, {len(batch)} entries dropped: {exc}", file=sys.stderr) + return False + return True async def close_async(self) -> None: """Flush all remaining entries and close the handler.""" - while self._queue: - await self._flush() + pending = len(self._queue) + while self._queue and pending > 0: + if not await self._flush(): + break + pending -= _BATCH_SIZE self.close()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry.py` around lines 154 - 158, Update close_async to guarantee bounded shutdown by stopping the drain after the first failed _flush, using a success/failure result from _flush to distinguish outcomes. Preserve flushing queued batches while delivery succeeds, then close the handler even when the queue remains or emit refills it.controller/internal/service/telemetry_service.go (1)
185-187: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winThe server is still plaintext while the controller advertises a CA certificate.
grpc.NewServer()is created with nogrpc.Creds(...).ControllerService.GetServiceEndpointsreturnsTelemetryConfig.Certificate, and_setup_telemetryinpython/packages/jumpstarter/jumpstarter/exporter/exporter.py(Lines 449-455) buildsgrpc.aio.secure_channelwheneverep.certificateis set. In that configuration the TLS handshake fails and no logs reach this service.Either configure server credentials here from the same certificate and its key, or do not advertise a certificate for this endpoint.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controller/internal/service/telemetry_service.go` around lines 185 - 187, The telemetry gRPC server created in the service startup path must match the certificate advertised by ControllerService.GetServiceEndpoints. Configure grpc.NewServer with TLS credentials built from the corresponding certificate and private key before registering TelemetryServiceServer, or stop returning TelemetryConfig.Certificate for this endpoint so clients use plaintext consistently.
🧹 Nitpick comments (1)
controller/internal/service/telemetry_service_test.go (1)
85-101: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest the real handler instead of a copy of its logic.
buildTelemetryEndpointsResponsereimplements the body ofControllerService.GetServiceEndpoints. The tests at Lines 103-154 assert on this copy, so they cannot detect a regression in the production handler. For example, if the"info"default is removed fromGetServiceEndpoints,TestGetServiceEndpoints_DefaultsMinSeverityToInfostill passes.Add a passing authenticator fake and call
svc.GetServiceEndpointsdirectly, in the same waytelemetrySvcWithConfigbuilds a failing one.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controller/internal/service/telemetry_service_test.go` around lines 85 - 101, The tests currently exercise the duplicate buildTelemetryEndpointsResponse helper instead of production behavior. Remove that helper and update the tests to construct a ControllerService through telemetrySvcWithConfig using a passing authenticator fake, then call svc.GetServiceEndpoints directly; preserve the existing assertions, including the default "info" severity case.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controller/internal/service/telemetry_service.go`:
- Around line 75-103: In the telemetry request handler around the subject
parsing and entries loop, reject any token whose subject is not exactly exporter
form before processing entries, returning PermissionDenied; do not allow empty
claimedName or claimedNamespace to bypass ownership checks. Also avoid
partial-batch failure by counting mismatched entries in dropped and continuing,
while stamping or validating each accepted entry’s Exporter and Namespace
against the token claims.
In `@protocol/proto/jumpstarter/v1/telemetry.proto`:
- Around line 40-41: Update PushLogs to prevent extra_fields from overriding
sink-owned telemetry identity fields such as component, exporter, severity, and
namespace by rejecting reserved keys or prefixing accepted keys before emission.
Preserve trusted field values in all output paths, and add a test covering
duplicate reserved keys.
---
Duplicate comments:
In `@controller/internal/service/telemetry_service.go`:
- Around line 185-187: The telemetry gRPC server created in the service startup
path must match the certificate advertised by
ControllerService.GetServiceEndpoints. Configure grpc.NewServer with TLS
credentials built from the corresponding certificate and private key before
registering TelemetryServiceServer, or stop returning
TelemetryConfig.Certificate for this endpoint so clients use plaintext
consistently.
In `@python/packages/jumpstarter/jumpstarter/exporter/telemetry.py`:
- Around line 154-158: Update close_async to guarantee bounded shutdown by
stopping the drain after the first failed _flush, using a success/failure result
from _flush to distinguish outcomes. Preserve flushing queued batches while
delivery succeeds, then close the handler even when the queue remains or emit
refills it.
---
Nitpick comments:
In `@controller/internal/service/telemetry_service_test.go`:
- Around line 85-101: The tests currently exercise the duplicate
buildTelemetryEndpointsResponse helper instead of production behavior. Remove
that helper and update the tests to construct a ControllerService through
telemetrySvcWithConfig using a passing authenticator fake, then call
svc.GetServiceEndpoints directly; preserve the existing assertions, including
the default "info" severity case.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: e1cc0689-349c-498d-bbf4-a7e4d1fdcbfa
⛔ Files ignored due to path filters (4)
controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/telemetry.pb.gois excluded by!**/*.pb.gocontroller/internal/protocol/jumpstarter/v1/telemetry_grpc.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (25)
controller/cmd/main.gocontroller/cmd/telemetry/main.gocontroller/internal/config/config.gocontroller/internal/config/types.gocontroller/internal/oidc/op.gocontroller/internal/service/controller_service.gocontroller/internal/service/telemetry_service.gocontroller/internal/service/telemetry_service_test.goprotocol/proto/jumpstarter/v1/jumpstarter.protoprotocol/proto/jumpstarter/v1/telemetry.protopython/packages/jumpstarter-protocol/jumpstarter_protocol/__init__.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/client/v1/client_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pyipython/packages/jumpstarter/jumpstarter/config/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/exporter_telemetry_test.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
🚧 Files skipped from review as they are similar to previous changes (12)
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/client/v1/client_pb2.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/init.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyi
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyi
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pyi
- python/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
- protocol/proto/jumpstarter/v1/jumpstarter.proto
- python/packages/jumpstarter/jumpstarter/exporter/exporter.py
- python/packages/jumpstarter/jumpstarter/exporter/exporter_telemetry_test.py
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controller/internal/service/telemetry_service.go`:
- Around line 57-67: The telemetry gRPC server started by Start does not
configure TLS despite advertising TelemetryConfig.Certificate to secure
exporters. Add certificate and private-key fields to TelemetryService, load and
validate them with tls.LoadX509KeyPair, construct TLS credentials, and pass
grpc.Creds(credentials.NewTLS(...)) to grpc.NewServer; preserve the existing
BindAddr and Signer behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: a569cb00-de58-48cc-b054-19aef42cd9e9
⛔ Files ignored due to path filters (1)
controller/internal/protocol/jumpstarter/v1/telemetry.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (16)
controller/cmd/main.gocontroller/cmd/telemetry/main.gocontroller/internal/config/config.gocontroller/internal/config/types.gocontroller/internal/oidc/op.gocontroller/internal/service/controller_service.gocontroller/internal/service/telemetry_service.gocontroller/internal/service/telemetry_service_test.goprotocol/proto/jumpstarter/v1/telemetry.protopython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pypython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pyipython/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pyipython/packages/jumpstarter/jumpstarter/config/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/exporter.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry.pypython/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
🚧 Files skipped from review as they are similar to previous changes (12)
- protocol/proto/jumpstarter/v1/telemetry.proto
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2_grpc.pyi
- controller/cmd/telemetry/main.go
- controller/internal/oidc/op.go
- controller/cmd/main.go
- python/packages/jumpstarter/jumpstarter/exporter/telemetry_test.py
- controller/internal/service/controller_service.go
- controller/internal/config/config.go
- python/packages/jumpstarter/jumpstarter/config/exporter.py
- python/packages/jumpstarter/jumpstarter/exporter/exporter.py
- python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/telemetry_pb2.pyi
- python/packages/jumpstarter/jumpstarter/exporter/telemetry.py
|
@bkhizgiy sorry, some conflicts, I am reviewing in the meanwhile. |
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
Assisted-by: Claude Sonnet 4.6 Signed-off-by: Bella Khizgiyaev <bkhizgiy@redhat.com>
mangelajo
left a comment
mangelajo
left a comment
There was a problem hiding this comment.
Only comments for follow up, specially the TLS and config-passing to the jumpstarter-telemetry container/deployment.
|
|
||
| authenticator, prefix, router, option, provisioning, leasePolicy, | ||
| hiddenLabels, deprecatedLabels, err := config.LoadConfiguration( | ||
| cfg, err := config.LoadConfiguration( |
There was a problem hiding this comment.
Thank you!, it keep being tangential to every PR touching this, but in the end, there it goes!
| // When Enabled is true the controller advertises the telemetry endpoint to | ||
| // exporters and clients via GetServiceEndpoints so they can push logs without | ||
| // holding cluster credentials. | ||
| type Telemetry struct { |
There was a problem hiding this comment.
oh, I confused this with the operator config. :-/
This is the internal config we pass to the jumpstarter-controller, but anyway I think it may be ok, the controller needs to know if Telemetry is enabled/disabled, where is the endpoint, what will be the certificate, so it can announce it on the "Get telemetry endpoint" call.
:-D
phew :) I thought my confusion made it bad, it looks good actually, thanks Bella!
| svc := &service.TelemetryService{ | ||
| BindAddr: bindAddr, | ||
| Signer: signer, | ||
| } | ||
|
|
||
| errCh := make(chan error, 1) | ||
| go func() { | ||
| errCh <- svc.Start(ctx) |
There was a problem hiding this comment.
I was investigating how the main controller configured TLS and the endpoint name (not sure if it's needed here really.., may be for the gRPC server), the telemetry service should follow the same established pattern for consistency, but can be done in follow up:
The existing controller and router services receive their TLS certificates via Secret volume mounts (not inline ConfigMap strings), and their endpoint addresses via environment variables on the Deployment pod spec.
-
TLS: The operator should create (or reference) a TLS Secret for the telemetry service (either via cert-manager
CertificateCR or a user-providedspec.telemetry.tls.certSecret), mount it at/tls/, and setEXTERNAL_CERT_PEM=/tls/tls.crtandEXTERNAL_KEY_PEM=/tls/tls.keyenv vars — exactly likecreateControllerDeploymentandcreateRouterDeploymentdo today. ThenStart()should read those files and configuregrpc.Creds(credentials.NewTLS(...))on the server, falling back to a self-signed cert when the env vars are absent.
(let's not do plaintext in any case, that can be dangerous since the wire always carries authentication tokens, now for exporters, but later will be clients as well pushing logs). -
Endpoint address: The telemetry endpoint should be passed to the controller pod as an env var (e.g.
GRPC_TELEMETRY_ENDPOINT) rather than auto-derived from the namespace in the ConfigMap loader. The operator already knows the address because it creates the Service/Route/Ingress for it — same as it does forGRPC_ENDPOINTandGRPC_ROUTER_ENDPOINT.
(AI made, human touch afterwards :D )
|
@mangelajo Thanks for the review! I’ll merge this one and follow up with a separate PR for the TLS setup and passing the telemetry endpoint through the deployment config/env vars. |
3 participants
Exporters now push structured logs to an optional jumpstarter-telemetry service, giving operators a single place to query what happened during any lease , which pipeline ran it, what driver operation failed, and with what result.
What changed:
New proto RPCs - GetServiceEndpoints lets exporters discover the telemetry service after registration. TelemetryService.PushLogs receives batched structured log entries.
Controller config - A new telemetry section in the controller ConfigMap (endpoint, certificate, min_severity) tells authenticated exporters where to send logs.
jumpstarter-telemetry service - A standalone Go binary that receives log entries and writes them to structured stdout. Ready to forward to Loki in a future phase without any protocol changes.
TelemetryLogHandler (Python) - A logging.Handler attached to the root logger after registration. It pulls correlation fields from structlog contextvars (lease_id, client, exporter) and spec.context fields (build_id, pipeline, etc.), batches them, and flushes every 500ms via PushLogs. Telemetry failures are silently dropped — device operations are never blocked.
What you can do with it:
A single log query like {lease="telemetry-test-4"} returns every event from that lease's full lifecycle. {pipeline="ci-main"} finds all leases across any run tagged with that pipeline. The correlation data answers questions like "which pipeline was running on this lease?" and "what driver operation failed?" without joining multiple sources.