Add support for shutdown_ms option

lib/vector/agent.ex

@@ -77,6 +77,10 @@ defmodule Vector.Agent do
     {:stop, :normal, agent}
   end
 
+  def handle_info(:shutdown, agent) do
+    {:stop, :normal, agent}
+  end
+
   @impl GenServer
   def terminate({[:vector, :error], {:exit_status, status}}, agent) do
     :ok = Logger.log(agent, :error, "vector: Vector is exiting with error status #{status}.")
@@ -85,6 +89,7 @@ defmodule Vector.Agent do
 
   def terminate(message, agent) when message in [:normal, :shutdown] do
     :ok = :exec.stop(agent.os_pid)
+    do_confirm_exit(agent.os_pid)
     :ok = Logger.log(agent, :info, "vector: Vector is stopping.")
     agent
   end
@@ -99,6 +104,7 @@ defmodule Vector.Agent do
     {:ok, pid, os_pid} = :exec.run_link(command, options)
     agent = %__MODULE__{config: config, pid: pid, os_pid: os_pid}
     :ok = Logger.log(agent, :info, "vector: Vector is starting.")
+    maybe_schedule_shutdown(agent)
     agent
   end
 
@@ -120,6 +126,30 @@ defmodule Vector.Agent do
     consumer.handle_data(agent, type, data, opts)
   end
 
+  defp maybe_schedule_shutdown(agent) do
+    if is_integer(agent.config.shutdown_ms) do
+      :ok =
+        Logger.log(
+          agent,
+          :info,
+          "vector: Vector shutdown scheduled in #{agent.config.shutdown_ms}ms."
+        )
+
+      Process.send_after(self(), :shutdown, agent.config.shutdown_ms)
+    end
+  end
+
+  defp do_confirm_exit(os_pid) do
+    case System.cmd("ps", ["-p #{os_pid}"]) do
+      {_, 0} ->
+        :timer.sleep(250)
+        do_confirm_exit(os_pid)
+
+      {_, 1} ->
+        :ok
+    end
+  end
+
   defimpl Inspect do
     def inspect(%{os_pid: os_pid}, _) do
       "#Vector.Agent<pid: #{inspect(os_pid)}>"

lib/vector/config.ex

@@ -20,6 +20,7 @@ defmodule Vector.Config do
     :allow_empty_config,
     :strict_env_vars,
     :log_level,
+    :shutdown_ms,
     start_async: true,
     stderr: {Vector.Consumer.Logger, []}
   ]
@@ -45,6 +46,7 @@ defmodule Vector.Config do
           allow_empty_config: boolean() | nil,
           strict_env_vars: boolean() | nil,
           log_level: :trace | :debug | :info | :warn | :error | :none | nil,
+          shutdown_ms: non_neg_integer() | nil,
           start_async: boolean()
         }
 

test/vector_test.exs

@@ -74,6 +74,38 @@ defmodule VectorTest do
     assert stdout == "foo\n"
   end
 
+  test "will shutdown based on shutdown_ms" do
+    config = %{
+      sources: %{
+        test_source: %{
+          type: "demo_logs",
+          format: "apache_common"
+        }
+      },
+      sinks: %{
+        test_sink: %{
+          type: "console",
+          inputs: ["test_source"],
+          encoding: %{
+            codec: "text"
+          }
+        }
+      }
+    }
+
+    config = %Vector.Config{
+      config: to_json_file!(config),
+      shutdown_ms: 250
+    }
+
+    assert {:ok, pid} = Vector.start_link(config)
+    assert Process.alive?(pid)
+
+    :timer.sleep(1500)
+
+    refute Process.alive?(pid)
+  end
+
   defp to_json_file!(config) do
     file = Enum.random(1..100_000)
     file = System.tmp_dir!() <> "/#{file}.json"