Refactor: Move StopTimeUpdate consumer into a new pipeline #493

arkadyan · 2020-03-05T23:10:51Z

Asana ticket: Refactor: Move StopTimeUpdate consumer into a new pipeline

paulswartz

some small comments before I hop on a plane

lib/concentrate/pipeline/stop_time_updates_pipeline.ex

lib/concentrate/pipeline.ex

skyqrose · 2020-03-09T14:21:38Z

lib/concentrate/pipeline/stop_time_updates_pipeline.ex

+  def source(opts) do
+    Pipeline.source(
+      :trip_updates_enhanced,
+      opts[:trip_updates_url],
+      Concentrate.Parser.GTFSRealtimeEnhanced
+    )
+  end


Will this behave well if the trip_updates_url isn't set?

No. I was thinking that isn't something we have a need to worry about—is there something I'm missing?

It might cause problems to any non-MBTA users since they won't be able to give a trip_updates_url that has the format we expect, but wouldn't make a difference to us.

skyqrose · 2020-03-09T14:24:48Z

lib/concentrate/supervisor.ex

+    Enum.concat([
+      Concentrate.Pipeline.VehiclePositionsPipeline.pipeline(opts),
+      Concentrate.Pipeline.StopTimeUpdatesPipeline.pipeline(opts)
+    ])


I'm thinking of the stop times pipeline as being pretty independent from the concentrate pipeline. What do you think about giving them separate supervisors to reflect that, and so if one pipeline dies, the other one doesn't have to get restarted with it. (I'm imagining both pipeline-supervisors would be started by Realtime.Supervisor)

Added individual supervisors for each pipeline.

👍 They can be restarted separately, but I think the structure is more complex than necessary.
It's now

Realtime.Supervisor ├ ... └ Concentrate.Supervisor ├ Concentrate.Pipeline.VehiclePositionsPipelineSupervisor │ └ Concentrate.Pipeline.VehiclePositionsPipeline │ ├ Concentrate.Producer.HTTP │ ├ Concentrate.Producer.HTTP │ ├ Concentrate.Merge │ └ Concentrate.Consumer.VehiclePositions └ Concentrate.Pipeline.StopTimeUpdatesPipelineSupervisor └ Concentrate.Pipeline.StopTimeUpdatesPipeline ├ Concentrate.Producer.HTTP └ Concentrate.Consumer.StopTimeUpdates

but we can knock out a couple layers like this:

Realtime.Supervisor ├ ... ├ Concentrate.VehiclePositionsPipeline │ ├ Concentrate.Producer.HTTP │ ├ Concentrate.Producer.HTTP │ ├ Concentrate.Merge │ └ Concentrate.Consumer.VehiclePositions └ Concentrate.Pipeline.StopTimeUpdatesPipeline ├ Concentrate.Producer.HTTP └ Concentrate.Consumer.StopTimeUpdates

That was intentional to encapsulate all the pipeline configuration knowledge. It didn't seem like something the Realtime supervisor should be concerned with.

Realtime.Supervisor wouldn't know anything about how the pipelines are set up. It would just know that there's 2 separate pipelines, one for VPs and one for STUs.

All the knowledge about which children to start and which env vars to use would be written in VehiclePositionsPipeline or StopTimeUpdatesPipeline.

Yes, knowing about what pipelines exist is what I wanted to encapsulate.

codecov · 2020-03-09T16:26:29Z

Codecov Report

Merging #493 into master will increase coverage by 0.18%.
The diff coverage is 100%.

@@            Coverage Diff            @@
##           master    #493      +/-   ##
=========================================
+ Coverage   97.51%   97.7%   +0.18%     
=========================================
  Files         166     169       +3     
  Lines        3747    3750       +3     
  Branches      516     516              
=========================================
+ Hits         3654    3664      +10     
+ Misses         90      83       -7     
  Partials        3       3

Impacted Files	Coverage Δ
lib/concentrate/consumer/stop_time_updates.ex	`100% <100%> (ø)`	⬆️
...concentrate/pipeline/stop_time_updates_pipeline.ex	`100% <100%> (ø)`
lib/concentrate/pipeline.ex	`100% <100%> (ø)`
...concentrate/pipeline/vehicle_positions_pipeline.ex	`100% <100%> (ø)`
lib/concentrate/supervisor.ex	`100% <100%> (+9.09%)`	⬆️
lib/concentrate/merge.ex	`89.28% <0%> (+5.35%)`	⬆️
lib/concentrate/consumer/vehicle_positions.ex	`100% <0%> (+14.28%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 9b1e385...223078d. Read the comment docs.

lib/concentrate/pipeline/stop_time_updates_pipeline.ex

paulswartz · 2020-03-09T17:48:43Z

lib/concentrate/pipeline/stop_time_updates_pipeline_supervisor.ex

+
+  @impl true
+  def init(opts) do
+    Supervisor.init(Concentrate.Pipeline.StopTimeUpdatesPipeline.init(opts), strategy: :rest_for_one)


This is also something like you could include in Pipeline, since you can pass a list of children to Supervisor.start_link. The two separate pipelines would be configured like this:

{Concentrate.Pipeline, module: Concentrate.Pipeline.StopTimeUpdates, ..other_opts..}, {Concentrate.Pipeline, module: Concentrate.Pipeline.VehiclePositions, ..other opts..}

The init/1 callback for Pipeline behaviors is the same, and then you implement start_link/1 and child_spec/1 on the Pipeline module directly. This is also a case where you'll want an id on the child spec: you can probably use the module for the ID.

I feel like I might not be fully grasping what you are saying here, perhaps starting with the bit about "you can pass a list of children to Supervisor.start_link". Are you suggesting something counter to the suggestion Sky made here of separating out the pipelines from a single supervisor? Or is what you are saying different?

I'll write it up more thoroughly tomorrow morning.

Thanks! We can also talk through it in person if that's easier.

defmodule Pipeline do @moduledoc """Create a pipeline supervisor for a given module.""" @callback init(Keyword.t) :: [Supervisor.child_spec] def start_link(mod, opts) do children = mod.init(opts) Supervisor.start_link(children, restart: :rest_for_one) end def child_spec(opts) do mod = Keyword.fetch(opts, :module) %{ type: :supervisor, id: mod, start: {__MODULE__, :start_link, [mod, opts]} } end end # in Concentrate.Supervisor def start_link(...) do Supervisor.start_link([ {Pipeline, module: StopTimePipeline, ...}, {Pipeline, module: VehiclePositionSupervisor, ...} ], restart: :one_for_one) end

Thanks, that's helpful. Let me know if you have any suggestions on my adaptation.

paulswartz

looking good!

paulswartz · 2020-03-10T16:07:21Z

lib/concentrate/pipeline/stop_time_updates_pipeline.ex

+        ]
+
+  @impl Concentrate.Pipeline
+  @spec init(opts()) :: [Supervisor.child_spec()]


You alias Concentrate.Pipeline but then keep using Concentrate.Pipeline in most places (in both modules).

You don't need the @spec here, as it's implicitly the same as the @callback you're @impling.

skyqrose

I think it's good now, but I think we should wait for a ✅ from paul, too.

skyqrose · 2020-03-10T16:12:27Z

lib/concentrate/pipeline.ex

+    Supervisor.start_link(children, strategy: :rest_for_one)
+  end
+
+  def child_spec(opts) do


Does this need a @impl true to say it implements the Supervisor.child_spec callback? (Or would that get in the way of this being a behavior itself?)

(Same for Concentrate.Supervisor.child_spec which doesn't currently have an @impl true.)

I don't believe so because these aren't declaring a behaviour. I added a spec here though.

skyqrose · 2020-03-10T16:22:19Z

lib/realtime/supervisor.ex

+           swiftly_realtime_vehicles_url:
+             Application.get_env(:skate, :swiftly_realtime_vehicles_url),
+           trip_updates_url: Application.get_env(:skate, :trip_updates_url)
+         ]


Why did this extra nesting come up?

Oops, moved that to the child_spec where it should actually be.

arkadyan · 2020-03-10T19:06:38Z

@paulswartz does this PR look good to you at this point?

Split this functionality out from Concentrate.Supervisor.

Separate this out from the VehiclePositionsPipeline since we don't need the merge step and could run on a different cycle.

Rename the root function to init. Rename the existing module to be PipelineHelpers.

paulswartz

🍰

arkadyan requested review from paulswartz and skyqrose March 5, 2020 23:10

arkadyan assigned paulswartz and skyqrose Mar 5, 2020

paulswartz reviewed Mar 6, 2020

View reviewed changes

lib/concentrate/pipeline/stop_time_updates_pipeline.ex Show resolved Hide resolved

lib/concentrate/pipeline.ex Outdated Show resolved Hide resolved

skyqrose reviewed Mar 9, 2020

View reviewed changes

arkadyan force-pushed the mss-stu-pipeline branch from 92a735b to 24f18eb Compare March 9, 2020 16:23

paulswartz reviewed Mar 9, 2020

View reviewed changes

arkadyan force-pushed the mss-stu-pipeline branch 4 times, most recently from 088ebab to e822509 Compare March 10, 2020 14:22

paulswartz reviewed Mar 10, 2020

View reviewed changes

skyqrose approved these changes Mar 10, 2020

View reviewed changes

arkadyan added 11 commits March 10, 2020 15:22

Refactor: Add a new VehiclePositionsPipeline

59c59c4

Split this functionality out from Concentrate.Supervisor.

Refactor: Break out a new StopTimeUpdates pipeline

8c1deb9

Separate this out from the VehiclePositionsPipeline since we don't need the merge step and could run on a different cycle.

Define a Pipeline behaviour that our individual pipelines implement

722bc87

Rename the root function to init. Rename the existing module to be PipelineHelpers.

Remove id argument from consumer function

06ef798

Break out individual supervisors for each pipeline

a20fb83

Move helper functions into Pipeline module

e171ad1

Use Pipeline as a supervisor for each pipeline

ae3c478

Improve test coverage

00662fb

Remove unneeded specs, use aliases

b3fe407

Add a spec for child_spec

5acdbe9

Move list wrapper to child_spec declaration

223078d

arkadyan force-pushed the mss-stu-pipeline branch from 74212aa to 223078d Compare March 10, 2020 19:22

paulswartz approved these changes Mar 11, 2020

View reviewed changes

arkadyan merged commit 46042ed into master Mar 11, 2020

arkadyan deleted the mss-stu-pipeline branch March 11, 2020 14:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor: Move StopTimeUpdate consumer into a new pipeline #493

Refactor: Move StopTimeUpdate consumer into a new pipeline #493

arkadyan commented Mar 5, 2020

paulswartz left a comment

skyqrose Mar 9, 2020

arkadyan Mar 9, 2020

skyqrose Mar 9, 2020

skyqrose Mar 9, 2020

arkadyan Mar 9, 2020

skyqrose Mar 9, 2020

arkadyan Mar 9, 2020

skyqrose Mar 9, 2020

arkadyan Mar 10, 2020

codecov bot commented Mar 9, 2020 •

edited

Loading

paulswartz Mar 9, 2020

arkadyan Mar 9, 2020

paulswartz Mar 9, 2020

arkadyan Mar 9, 2020

paulswartz Mar 10, 2020

arkadyan Mar 10, 2020

paulswartz left a comment

paulswartz Mar 10, 2020

skyqrose left a comment

skyqrose Mar 10, 2020

arkadyan Mar 10, 2020 •

edited

Loading

skyqrose Mar 10, 2020

arkadyan Mar 10, 2020

arkadyan commented Mar 10, 2020

paulswartz left a comment

Refactor: Move StopTimeUpdate consumer into a new pipeline #493

Refactor: Move StopTimeUpdate consumer into a new pipeline #493

Conversation

arkadyan commented Mar 5, 2020

paulswartz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Mar 9, 2020 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

paulswartz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

skyqrose left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arkadyan Mar 10, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arkadyan commented Mar 10, 2020

paulswartz left a comment

Choose a reason for hiding this comment

codecov bot commented Mar 9, 2020 •

edited

Loading

arkadyan Mar 10, 2020 •

edited

Loading