Class: ActiveJob::Temporal::CredentialRefresher

Inherits:
Object
  • Object
show all
Defined in:
lib/activejob/temporal/credential_refresher.rb

Overview

Keeps credential files fresh in a long-lived process: TLS material, API key files, or anything else read once at boot and rotated underneath the process later.

Change detection is a content digest compared on an interval, so the only thing that can fire a callback is the bytes behind a path actually differing. Filesystem events (see :file_events) are an optional accelerator that wakes the loop early; they never decide whether a reload happens, so a missing listen gem or a mount without working inotify costs latency, not correctness.

Examples:

Worker: rebuild the client on cert rotation, patch the live connection on token rotation

refresher = CredentialRefresher.from_config(
  ActiveJob::Temporal.config,
  on_tls_change: -> { client_reloader.reload(source: "credential_refresh") }
).start

Enqueue-side process (web, Sidekiq) with a rotating projected token

require "activejob/temporal/credential_refresher"
ActiveJob::Temporal::CredentialRefresher.from_config(ActiveJob::Temporal.config).start

Defined Under Namespace

Classes: Source

Constant Summary collapse

DEFAULT_POLL_INTERVAL = 
30

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(sources:, poll_interval: DEFAULT_POLL_INTERVAL, file_events: false, logger: ActiveJob::Temporal::Logger, listener_factory: nil) ⇒ CredentialRefresher

Returns a new instance of CredentialRefresher.

Parameters:

  • sources (Array<Source>) —

    credential groups to keep fresh

  • poll_interval (Integer, Float) (defaults to: DEFAULT_POLL_INTERVAL) —

    seconds between digest comparisons

  • file_events (Boolean) (defaults to: false) —

    wake the loop early on filesystem events, requires listen

  • logger (#log_event, #warn, #error) (defaults to: ActiveJob::Temporal::Logger) —

    structured logger

  • listener_factory (#to, nil) (defaults to: nil) —

    injection point for tests, defaults to Listen



104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 104

def initialize(sources:, poll_interval: DEFAULT_POLL_INTERVAL, file_events: false,
               logger: ActiveJob::Temporal::Logger, listener_factory: nil)
  @sources = sources.reject { |source| source.paths.compact.empty? }
  @poll_interval = poll_interval
  @file_events = file_events
  @logger = logger
  @listener_factory = listener_factory
  @wakeups = Thread::Queue.new
  @digests = {}
  @thread = nil
  @listener = nil
end

Class Method Details

.from_config(configuration, on_tls_change: -> { ActiveJob::Temporal.reload_client! }, logger: ActiveJob::Temporal::Logger, listener_factory: nil) ⇒ CredentialRefresher

Note:

A process running a Temporalio::Worker MUST pass on_tls_change. The worker holds its own client reference, and the default reload_client! closes the client it replaces

  • which is the one the worker is still polling on. Pass a callback that assigns the fresh client to the worker, as WorkerClientReloader does.

Builds a refresher for every credential file the configuration names. A path is only configured because the credential behind it rotates, so credential_watch gates the whole mechanism rather than each file individually.

A token refresh mutates the live connection, so it is the same call in every process. A certificate refresh is not: the default rebuilds the memoized client, which is right for an enqueue-side process but wrong inside a worker.

Parameters:

  • configuration (Configuration) —

    the gem configuration

  • on_tls_change (#call) (defaults to: -> { ActiveJob::Temporal.reload_client! }) —

    invoked when any TLS file changes; see the note above

  • logger (#log_event, #warn, #error) (defaults to: ActiveJob::Temporal::Logger) —

    structured logger

  • listener_factory (#to, nil) (defaults to: nil) —

    injection point for tests, defaults to Listen

Returns:

  • (CredentialRefresher) —

    a refresher with no sources when no credential file is configured, or when credential_watch is off



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 62

def from_config(configuration,
                on_tls_change: -> { ActiveJob::Temporal.reload_client! },
                logger: ActiveJob::Temporal::Logger,
                listener_factory: nil)
  sources = []

  if configuration.credential_watch
    tls = tls_paths(configuration)
    sources << Source.new(name: "tls", paths: tls, on_change: on_tls_change) if tls.any?

    token = configuration.api_key_file.to_s.strip
    if token.present?
      sources << Source.new(name: "api_key", paths: [token],
                            on_change: -> { ActiveJob::Temporal.refresh_api_key! })
    end
  end

  new(
    sources: sources,
    poll_interval: configuration.credential_poll_interval,
    file_events: configuration.credential_file_events,
    logger: logger,
    listener_factory: listener_factory
  )
end

.tls_paths(configuration) ⇒ Array<String>

Returns configured TLS file paths, blanks removed.

Parameters:

Returns:

  • (Array<String>) —

    configured TLS file paths, blanks removed



90
91
92
93
94
95
96
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 90

def tls_paths(configuration)
  [
    configuration.tls_cert_path,
    configuration.tls_key_path,
    configuration.tls_server_root_ca_cert_path
  ].compact.reject { |path| path.to_s.strip.empty? }
end

Instance Method Details

#nudge ⇒ void

This method returns an undefined value.

Wakes the poll loop so the next comparison happens now instead of on the next tick. Decides nothing on its own.



151
152
153
154
155
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 151

def nudge
  @wakeups << :wakeup
rescue ClosedQueueError
  nil
end

#refresh_changed_sources ⇒ void

This method returns an undefined value.

Re-reads every source and fires the callbacks whose content changed. Safe to call at any rate: unchanged content is a no-op.



143
144
145
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 143

def refresh_changed_sources
  @sources.each { |source| refresh(source) }
end

#start ⇒ self

Baselines each digest before polling, so a process booting on an already-rotated file does not fire a redundant reload on its first check.

Returns:

  • (self) 


121
122
123
124
125
126
127
128
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 121

def start
  return self if @sources.empty? || @thread

  @sources.each { |source| @digests[source.name] = digest(source) }
  @thread = Thread.new { run }
  start_listener if @file_events
  self
end

#stop ⇒ void

This method returns an undefined value.



131
132
133
134
135
136
137
 
# File 'lib/activejob/temporal/credential_refresher.rb', line 131

def stop
  @listener&.stop
  @listener = nil
  @wakeups.close
  @thread&.join
  @thread = nil
end