feat(subscriber) expose server parts #451

hds · 2023-07-18T15:13:09Z

The ConsoleLayer builder provides the user with a console layer and
a server, which is used to start the gRPC server.

However, it may be desireable to expose the instrumentation server together
with other services on the same Tonic router. This was requested
explicitly in #449 428.

Additionally, to add tests which make use of the instrumentation server
(as part of improving test coverage for #450), more flexibility is
needed than what is provided by the current API. Specifically we would
like to connect a client and server via an in memory channel, rather
than a TCP connection.

This change adds an additional method to console_subscriber::Server
called into_parts which allows the user to access the
InstrumentServer directly. The Aggregator is also returned and
must be set to run for at least as long as the instrument server. This
allows the aggregator to be spawned wherever the user wishes.

To facilitate the addition of functionality which would result in more
"parts" in the future, into_parts returns a non-exhaustive struct,
rather than a tuple of parts.

Closes: #428

The `ConsoleLayer` builder provides the user with a console layer and a server, which is used to start the gRPC server. However, it may be desireable to expose the instrumentation server together with other services on the same Tonic router. This was requested explicitly in #428. Additionally, to add tests which make use of the instrumentation server (as part of improving test coverage for #450), more flexibility is needed than what is provided by the current API. Specifically we would like to connect a client and server via an in memory channel, rather than a TCP connection. This change adds an additional method to `console_subscriber::Server` called `into_parts` which allows the user to access the `InstrumentServer` directly. A handle which controls the lifetime of the `Aggregator` is also provided, as the user must ensure that the aggregator lives at least as long as the instrument server. To facilitate the addition of functionality which would result in more "parts" in the future, `into_parts` returns a non-exhaustive struct, rather than a tuple of parts. Closes: #428

The `console-subscriber` crate has no integration tests. There are some unit tests, but without very high coverage of features. Recently, we've found or fixed a few errors which probably could have been caught by a medium level of integration testing. However, testing `console-subscriber` isn't straight forward. It is effectively a tracing subscriber (or layer) on one end, and a gRPC server on the other end. This change adds enough of a testing framework to write some initial integration tests. Each test comprises 2 parts: - One or more "expcted tasks" - A future which will be driven to completion on a dedicated Tokio runtime. Behind the scenes, a console subscriber layer is created and it's server part is connected to a duplex stream. The client of the duplex stream then records incoming updates and reconstructs "actual tasks". The layer itself is set as the default subscriber for the duration of `block_on` which is used to drive the provided future to completioin. The expected tasks have a set of "matches", which is how we find the actual task that we want to validate against. Currently, the only value we match on is the task's name. The expected tasks also have a set of expectations. These are other fields on the actual task which are validated once a matching task is found. Currently, the two fields which can have expectations set on them are the `wakes` and `self_wakes` fields. So, to construct an expected task, which will match a task with the name `"my-task"` and then validate that the matched task gets woken once, the code would be: ```rust ExpectedTask::default() .match_name("my-task") .expect_wakes(1); ``` A future which passes this test could be: ```rust async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) } ``` The full test would then look like: ```rust fn wakes_once() { let expected_task = ExpectedTask::default() .match_name("my-task") .expect_wakes(1); let future = async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) }; assert_task(expected_task, future); } ``` The PR depends on 2 others: - #447 which fixes an error in the logic that determines whether a task is retained in the aggregator or not. - #451 which exposes the server parts and is necessary to allow us to connect the instrument server and client via a duplex channel. This change contains some initial tests for wakes and self wakes which would have caught the error fixed in #430. Additionally there are tests for the functionality of the testing framework itself.

The `console-subscriber` crate has no integration tests. There are some unit tests, but without very high coverage of features. Recently, we've found or fixed a few errors which probably could have been caught by a medium level of integration testing. However, testing `console-subscriber` isn't straight forward. It is effectively a tracing subscriber (or layer) on one end, and a gRPC server on the other end. This change adds enough of a testing framework to write some initial integration tests. It is the first step towards closing #450. Each test comprises 2 parts: - One or more "expcted tasks" - A future which will be driven to completion on a dedicated Tokio runtime. Behind the scenes, a console subscriber layer is created and it's server part is connected to a duplex stream. The client of the duplex stream then records incoming updates and reconstructs "actual tasks". The layer itself is set as the default subscriber for the duration of `block_on` which is used to drive the provided future to completioin. The expected tasks have a set of "matches", which is how we find the actual task that we want to validate against. Currently, the only value we match on is the task's name. The expected tasks also have a set of expectations. These are other fields on the actual task which are validated once a matching task is found. Currently, the two fields which can have expectations set on them are the `wakes` and `self_wakes` fields. So, to construct an expected task, which will match a task with the name `"my-task"` and then validate that the matched task gets woken once, the code would be: ```rust ExpectedTask::default() .match_name("my-task") .expect_wakes(1); ``` A future which passes this test could be: ```rust async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) } ``` The full test would then look like: ```rust fn wakes_once() { let expected_task = ExpectedTask::default() .match_name("my-task") .expect_wakes(1); let future = async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) }; assert_task(expected_task, future); } ``` The PR depends on 2 others: - #447 which fixes an error in the logic that determines whether a task is retained in the aggregator or not. - #451 which exposes the server parts and is necessary to allow us to connect the instrument server and client via a duplex channel. This change contains some initial tests for wakes and self wakes which would have caught the error fixed in #430. Additionally there are tests for the functionality of the testing framework itself.

console-subscriber/src/lib.rs

Co-authored-by: Eliza Weisman <eliza@buoyant.io>

Instead of aborting the aggregator task upon dropping the `AggregatorHandle`, we now provide an explicit `abort` method. If the user chooses, they can discard the aggregator handle and while they will lose the possibility to clean up the aggregator task, they will be otherwise unaffected.

As discussed in code review, for such a low level api as `Server::into_parts`, it makes sense to allow the user to spawn the aggregator where they like, rather than spawning it internally. Since I couldn't find a way to return and use a `dyn Future` (boxed or otherwise), the `Aggregator` has been made public with a (async) single function `run()` which will start it's run loop.

hawkw

Thanks for going along with my last-minute suggested changes, this looks great!

The `console-subscriber` crate has no integration tests. There are some unit tests, but without very high coverage of features. Recently, we've found or fixed a few errors which probably could have been caught by a medium level of integration testing. However, testing `console-subscriber` isn't straight forward. It is effectively a tracing subscriber (or layer) on one end, and a gRPC server on the other end. This change adds enough of a testing framework to write some initial integration tests. It is the first step towards closing #450. Each test comprises 2 parts: - One or more "expcted tasks" - A future which will be driven to completion on a dedicated Tokio runtime. Behind the scenes, a console subscriber layer is created and it's server part is connected to a duplex stream. The client of the duplex stream then records incoming updates and reconstructs "actual tasks". The layer itself is set as the default subscriber for the duration of `block_on` which is used to drive the provided future to completioin. The expected tasks have a set of "matches", which is how we find the actual task that we want to validate against. Currently, the only value we match on is the task's name. The expected tasks also have a set of expectations. These are other fields on the actual task which are validated once a matching task is found. Currently, the two fields which can have expectations set on them are the `wakes` and `self_wakes` fields. So, to construct an expected task, which will match a task with the name `"my-task"` and then validate that the matched task gets woken once, the code would be: ```rust ExpectedTask::default() .match_name("my-task") .expect_wakes(1); ``` A future which passes this test could be: ```rust async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) } ``` The full test would then look like: ```rust fn wakes_once() { let expected_task = ExpectedTask::default() .match_name("my-task") .expect_wakes(1); let future = async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) }; assert_task(expected_task, future); } ``` The PR depends on 2 others: - #447 which fixes an error in the logic that determines whether a task is retained in the aggregator or not. - #451 which exposes the server parts and is necessary to allow us to connect the instrument server and client via a duplex channel. This change contains some initial tests for wakes and self wakes which would have caught the error fixed in #430. Additionally there are tests for the functionality of the testing framework itself.

The `console-subscriber` crate has no integration tests. There are some unit tests, but without very high coverage of features. Recently, we've found or fixed a few errors which probably could have been caught by a medium level of integration testing. However, testing `console-subscriber` isn't straight forward. It is effectively a tracing subscriber (or layer) on one end, and a gRPC server on the other end. This change adds enough of a testing framework to write some initial integration tests. It is the first step towards closing #450. Each test comprises 2 parts: - One or more "expected tasks" - A future which will be driven to completion on a dedicated Tokio runtime. Behind the scenes, a console subscriber layer is created and its server part is connected to a duplex stream. The client of the duplex stream then records incoming updates and reconstructs "actual tasks". The layer itself is set as the default subscriber for the duration of `block_on` which is used to drive the provided future to completioin. The expected tasks have a set of "matches", which is how we find the actual task that we want to validate against. Currently, the only value we match on is the task's name. The expected tasks also have a set of "expectations". These are other fields on the actual task which are validated once a matching task is found. Currently, the two fields which can have expectations set on them are `wakes` and `self_wakes`. So, to construct an expected task, which will match a task with the name `"my-task"` and then validate that the matched task gets woken once, the code would be: ```rust ExpectedTask::default() .match_name("my-task") .expect_wakes(1); ``` A future which passes this test could be: ```rust async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) } ``` The full test would then look like: ```rust fn wakes_once() { let expected_task = ExpectedTask::default() .match_name("my-task") .expect_wakes(1); let future = async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) }; assert_task(expected_task, future); } ``` The PR depends on 2 others: - #447 which fixes an error in the logic that determines whether a task is retained in the aggregator or not. - #451 which exposes the server parts and is necessary to allow us to connect the instrument server and client via a duplex channel. This change contains some initial tests for wakes and self wakes which would have caught the error fixed in #430. Additionally there are tests for the functionality of the testing framework itself. Co-authored-by: Eliza Weisman <eliza@buoyant.io>

The `ConsoleLayer` builder provides the user with a console layer and a server, which is used to start the gRPC server. However, it may be desireable to expose the instrumentation server together with other services on the same Tonic router. This was requested explicitly in #449 428. Additionally, to add tests which make use of the instrumentation server (as part of improving test coverage for #450), more flexibility is needed than what is provided by the current API. Specifically we would like to connect a client and server via an in memory channel, rather than a TCP connection. This change adds an additional method to `console_subscriber::Server` called `into_parts` which allows the user to access the `InstrumentServer` directly. The `Aggregator` is also returned and must be set to run for at least as long as the instrument server. This allows the aggregator to be spawned wherever the user wishes. To facilitate the addition of functionality which would result in more "parts" in the future, `into_parts` returns a non-exhaustive struct, rather than a tuple of parts. Closes: #428

The `console-subscriber` crate has no integration tests. There are some unit tests, but without very high coverage of features. Recently, we've found or fixed a few errors which probably could have been caught by a medium level of integration testing. However, testing `console-subscriber` isn't straight forward. It is effectively a tracing subscriber (or layer) on one end, and a gRPC server on the other end. This change adds enough of a testing framework to write some initial integration tests. It is the first step towards closing #450. Each test comprises 2 parts: - One or more "expected tasks" - A future which will be driven to completion on a dedicated Tokio runtime. Behind the scenes, a console subscriber layer is created and its server part is connected to a duplex stream. The client of the duplex stream then records incoming updates and reconstructs "actual tasks". The layer itself is set as the default subscriber for the duration of `block_on` which is used to drive the provided future to completioin. The expected tasks have a set of "matches", which is how we find the actual task that we want to validate against. Currently, the only value we match on is the task's name. The expected tasks also have a set of "expectations". These are other fields on the actual task which are validated once a matching task is found. Currently, the two fields which can have expectations set on them are `wakes` and `self_wakes`. So, to construct an expected task, which will match a task with the name `"my-task"` and then validate that the matched task gets woken once, the code would be: ```rust ExpectedTask::default() .match_name("my-task") .expect_wakes(1); ``` A future which passes this test could be: ```rust async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) } ``` The full test would then look like: ```rust fn wakes_once() { let expected_task = ExpectedTask::default() .match_name("my-task") .expect_wakes(1); let future = async { task::Builder::new() .name("my-task") .spawn(async { tokio::time::sleep(std::time::Duration::ZERO).await }) }; assert_task(expected_task, future); } ``` The PR depends on 2 others: - #447 which fixes an error in the logic that determines whether a task is retained in the aggregator or not. - #451 which exposes the server parts and is necessary to allow us to connect the instrument server and client via a duplex channel. This change contains some initial tests for wakes and self wakes which would have caught the error fixed in #430. Additionally there are tests for the functionality of the testing framework itself. Co-authored-by: Eliza Weisman <eliza@buoyant.io>

# Changelog All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## console-subscriber-v0.2.0 - (2023-09-29) [0b0c1af](https://github.com/tokio-rs/console/commit/0b0c1aff18c3260d3a45a78f6c0d6f4206af1cbb)...[0b0c1af](https://github.com/tokio-rs/console/commit/0b0c1aff18c3260d3a45a78f6c0d6f4206af1cbb) ### <a id = "console-subscriber-v0.2.0-breaking"></a>Breaking Changes - **Update Tonic and Prost dependencies ([#364](#364 ([f9b8e03](https://github.com/tokio-rs/console/commit/f9b8e03bd7ee1d0edb441c94a93a350d5b06ed3b))<br />This commit updates the public dependencies `prost` and `tonic` to semver-incompatible versions (v0.11.0 and v0.8.0, respectively). This is a breaking change for users who are integrating the `console-api` protos with their own `tonic` servers or clients. - **Update `tonic` to v0.10 and increase MSRV to 1.64 ([#464](#464 ([96e62c8](https://github.com/tokio-rs/console/commit/96e62c83ef959569bb062dc8fee98fa2b2461e8d))<br />This is a breaking change for users of `console-api` and `console-subscriber`, as it changes the public `tonic` dependency to a semver-incompatible version. This breaks compatibility with `tonic` 0.9.x and `prost` 0.11.x. ### Added - [**breaking**](#console-subscriber-v0.2.0-breaking) Update Tonic and Prost dependencies ([#364](#364)) ([f9b8e03](f9b8e03)) - Add support for Unix domain sockets ([#388](#388)) ([a944dbc](a944dbc), closes [#296](#296)) - Add scheduled time per task ([#406](#406)) ([f280df9](f280df9)) - Add task scheduled times histogram ([#409](#409)) ([d92a399](d92a399)) - Update `tonic` to 0.9 ([#420](#420)) ([48af1ee](48af1ee)) - Update MSRV to Rust 1.60.0 ([b18ee47](b18ee47)) - Expose server parts ([#451](#451)) ([e51ac5a](e51ac5a)) - Add cfg `console_without_tokio_unstable` ([#446](#446)) ([7ed6673](7ed6673)) - Add warning for tasks that never yield ([#439](#439)) ([d05fa9e](d05fa9e)) - [**breaking**](#console-subscriber-v0.2.0-breaking) Update `tonic` to v0.10 and increase MSRV to 1.64 ([#464](#464)) ([96e62c8](96e62c8)) ### Documented - Fix unclosed code block ([#463](#463)) ([362bdbe](362bdbe)) - Update MSRV version docs to 1.64 ([#467](#467)) ([94a5a51](94a5a51)) ### Fixed - Fix build on tokio 1.21.0 ([#374](#374)) ([c34ac2d](c34ac2d)) - Fix off-by-one indexing for `callsites` ([#391](#391)) ([43891ab](43891ab)) - Bump minimum Tokio version ([#397](#397)) ([bbb8f25](bbb8f25), fixes [#386](#386)) - Fix self wakes count ([#430](#430)) ([d308935](d308935)) - Remove clock skew warning in `start_poll` ([#434](#434)) ([4a88b28](4a88b28)) - Do not report excessive polling ([#378](#378)) ([#440](#440)) ([8b483bf](8b483bf), closes [#378](#378)) - Correct retain logic ([#447](#447)) ([36ffc51](36ffc51)) Signed-off-by: Eliza Weisman <eliza@buoyant.io>

A number of new or updated Clippy lints in Rust 1.80.0 need to be fixed. One of them pointed out that the `AggregatorHandle` is not used, and is also not constructable from outside the crate. This struct was introduced in #451 as part of the `Server::into_parts` method. Originally, this method was going to return the `AggregatorHandle`, which wrapped the join handle from the task where the `Aggregator` had been spawned. This was later replaced by returning the `Aggregator` itself, which the user had the obligation to spawn themselves. However, it seems that the `AggregatorHandle` wasn't removed, even though it was never used. A new lint is the one for unexpected `--cfg` items. We now need to declare those in `Cargo.toml`. An update to `needless_borrows_for_generic_args` causes a false positive changing a `&mut` to a move, which we can't do as the same value is used afterwards.

A number of new or updated Clippy lints in Rust 1.80.0 need to be fixed. An update to the `dead_code` pointed out that the `AggregatorHandle` is not used, and it is not constructable from outside the crate because it has a private field. This struct was introduced in #451 as part of the `Server::into_parts` method. Originally, this method was going to return the `AggregatorHandle`, which wrapped the join handle from the task where the `Aggregator` had been spawned. This was later replaced by returning the `Aggregator` itself, which the user had the obligation to spawn themselves. However, it seems that the `AggregatorHandle` wasn't removed, even though it was never used. A new lint is the one for unexpected `--cfg` items. We now need to declare those in `Cargo.toml`. An update to `needless_borrows_for_generic_args` causes a false positive changing a `&mut` to a move, which we can't do as the same value is used afterwards.

…#578) A number of new or updated Clippy lints in Rust 1.80.0 need to be fixed. An update to the `dead_code` pointed out that the `AggregatorHandle` is not used, and it is not constructable from outside the crate because it has a private field. This struct was introduced in #451 as part of the `Server::into_parts` method. Originally, this method was going to return the `AggregatorHandle`, which wrapped the join handle from the task where the `Aggregator` had been spawned. This was later replaced by returning the `Aggregator` itself, which the user had the obligation to spawn themselves. However, it seems that the `AggregatorHandle` wasn't removed, even though it was never used. A new lint is the one for unexpected `--cfg` items. We now need to declare those in `Cargo.toml`. An update to `needless_borrows_for_generic_args` causes a false positive changing a `&mut` to a move, which we can't do as the same value is used afterwards.

hds requested a review from a team as a code owner July 18, 2023 15:13

hds force-pushed the hds/subscriber-server-into-parts branch from 93eb450 to 758ba9c Compare July 18, 2023 15:22

hds mentioned this pull request Jul 18, 2023

test(subscriber): add initial integration tests #452

Merged

hawkw reviewed Jul 18, 2023

View reviewed changes

hds and others added 4 commits July 19, 2023 12:34

Apply suggestions from code review

1d0213d

Co-authored-by: Eliza Weisman <eliza@buoyant.io>

fixed doc link

adca82c

hawkw approved these changes Jul 27, 2023

View reviewed changes

Merge branch 'main' into hds/subscriber-server-into-parts

86b35bb

hawkw merged commit f4536af into main Jul 28, 2023
11 checks passed

hawkw deleted the hds/subscriber-server-into-parts branch July 28, 2023 21:34

hds mentioned this pull request Jul 29, 2024

fix(subscriber): remove unused AggregatorHandle and fix other lints #578

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(subscriber) expose server parts #451

feat(subscriber) expose server parts #451

hds commented Jul 18, 2023 •

edited

Loading

hawkw left a comment

feat(subscriber) expose server parts #451

feat(subscriber) expose server parts #451

Conversation

hds commented Jul 18, 2023 • edited Loading

hawkw left a comment

Choose a reason for hiding this comment

hds commented Jul 18, 2023 •

edited

Loading