Class: Google::Cloud::Logging::AsyncWriter

Inherits:
Object
  • Object
show all
Defined in:
lib/google/cloud/logging/async_writer.rb

Overview

AsyncWriter

An object that batches and transmits log entries asynchronously.

Use this object to transmit log entries efficiently. It keeps a queue of log entries, and runs a background thread that transmits them to the logging service in batches. Generally, adding to the queue will not block.

This object is thread-safe; it may accept write requests from multiple threads simultaneously, and will serialize them when executing in the background thread.

Examples:

require "google/cloud/logging"

logging = Google::Cloud::Logging.new

async = logging.async_writer

entry1 = logging.entry payload: "Job started."
entry2 = logging.entry payload: "Job completed."

labels = { job_size: "large", job_code: "red" }
resource = logging.resource "gae_app",
                            "module_id" => "1",
                            "version_id" => "20150925t173233"

async.write_entries [entry1, entry2],
                    log_name: "my_app_log",
                    resource: resource,
                    labels: labels

Constant Summary collapse

DEFAULT_MAX_QUEUE_SIZE = 
10000
CLEANUP_TIMEOUT = 
10.0
WAIT_INTERVAL = 
1.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#last_exception ⇒ Object (readonly)

The last exception thrown by the background thread, or nil if nothing has been thrown.



93
94
95
 
# File 'lib/google/cloud/logging/async_writer.rb', line 93

def last_exception
  @last_exception
end

#state ⇒ Object (readonly)

The current state. Either :running, :suspended, :stopping, or :stopped



88
89
90
 
# File 'lib/google/cloud/logging/async_writer.rb', line 88

def state
  @state
end

Instance Method Details

#logger(log_name, resource, labels = {}) ⇒ Google::Cloud::Logging::Logger

Creates a logger instance that is API-compatible with Ruby's standard library Logger.

The logger will use AsyncWriter to transmit log entries on a background thread.

Examples:

require "google/cloud/logging"

logging = Google::Cloud::Logging.new

resource = logging.resource "gae_app",
                            module_id: "1",
                            version_id: "20150925t173233"

async = logging.async_writer
logger = async.logger "my_app_log", resource, env: :production
logger.info "Job started."

Parameters:

  • log_name (String) —

    A log resource name to be associated with the written log entries.

  • resource (Google::Cloud::Logging::Resource) —

    The monitored resource to be associated with written log entries.

  • labels (Hash) (defaults to: {}) —

    A set of user-defined data to be associated with written log entries.

Returns:



193
194
195
 
# File 'lib/google/cloud/logging/async_writer.rb', line 193

def logger log_name, resource, labels = {}
  Logger.new self, log_name, resource, labels
end

#resume ⇒ Boolean

Resumes this suspended asynchronous writer.

After this call succeeds, the state will change to :running, and the writer will resume sending RPCs.

Returns:

  • (Boolean) —

    Returns true if the writer had been suspended and is now running, otherwise false.



252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/google/cloud/logging/async_writer.rb', line 252

def resume
  ensure_thread
  @lock.synchronize do
    if state == :suspended
      @state = :running
      @lock_cond.broadcast
      true
    else
      false
    end
  end
end

#running? ⇒ Boolean

Returns true if this writer is running.

Returns:

  • (Boolean) —

    Returns true if the writer is currently running.



270
271
272
273
274
275
 
# File 'lib/google/cloud/logging/async_writer.rb', line 270

def running?
  ensure_thread
  @lock.synchronize do
    state == :running
  end
end

#stop ⇒ Boolean

Stops this asynchronous writer.

After this call succeeds, the state will change to :stopping, and you may not issue any additional write_entries calls. Any previously issued writes will complete. Once any existing backlog has been cleared, the state will change to :stopped.

Returns:

  • (Boolean) —

    Returns true if the writer was running, or false if the writer had already been stopped.



208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/google/cloud/logging/async_writer.rb', line 208

def stop
  ensure_thread
  @lock.synchronize do
    if state != :stopped
      @state = :stopping
      @lock_cond.broadcast
      true
    else
      false
    end
  end
end

#stop!(timeout, force: false) ⇒ Symbol

Stop this asynchronous writer and block until it has been stopped.

Parameters:

  • timeout (Number) —

    Timeout in seconds.

  • force (Boolean) —

    If set to true, and the writer hasn't stopped within the given timeout, kill it forcibly by terminating the thread. This should be used with extreme caution, as it can leave RPCs unfinished. Default is false.

Returns:

  • (Symbol) —

    Returns :stopped if the AsyncWriter was already stopped at the time of invocation, :waited if it stopped during the timeout period, :timeout if it is still running after the timeout, or :forced if it was forcibly killed.



353
354
355
356
357
358
359
360
 
# File 'lib/google/cloud/logging/async_writer.rb', line 353

def stop! timeout, force: false
  return :stopped unless stop
  return :waited if wait_until_stopped timeout
  return :timeout unless force
  @thread.kill
  @thread.join
  :forced
end

#stopped? ⇒ Boolean

Returns true if this writer is fully stopped.

Returns:

  • (Boolean) —

    Returns true if the writer is fully stopped.



307
308
309
310
311
312
 
# File 'lib/google/cloud/logging/async_writer.rb', line 307

def stopped?
  ensure_thread
  @lock.synchronize do
    state == :stopped
  end
end

#suspend ⇒ Boolean

Suspends this asynchronous writer.

After this call succeeds, the state will change to :suspended, and the writer will stop sending RPCs until resumed.

Returns:

  • (Boolean) —

    Returns true if the writer had been running and was suspended, otherwise false.



230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/google/cloud/logging/async_writer.rb', line 230

def suspend
  ensure_thread
  @lock.synchronize do
    if state == :running
      @state = :suspended
      @lock_cond.broadcast
      true
    else
      false
    end
  end
end

#suspended? ⇒ Boolean

Returns true if this writer is suspended.

Returns:

  • (Boolean) —

    Returns true if the writer is currently suspended.



282
283
284
285
286
287
 
# File 'lib/google/cloud/logging/async_writer.rb', line 282

def suspended?
  ensure_thread
  @lock.synchronize do
    state == :suspended
  end
end

#wait_until_stopped(timeout = nil) ⇒ Boolean

Blocks until this asynchronous writer has been stopped, or the given timeout (if present) has elapsed.

Parameters:

  • timeout (Number, nil) (defaults to: nil) —

    Timeout in seconds, or nil for no timeout.

Returns:

  • (Boolean) —

    Returns true if the writer is stopped, or false if the timeout expired.



324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/google/cloud/logging/async_writer.rb', line 324

def wait_until_stopped timeout = nil
  ensure_thread
  deadline = timeout ? ::Time.new.to_f + timeout : nil
  @lock.synchronize do
    until state == :stopped
      cur_time = ::Time.new.to_f
      return false if deadline && cur_time >= deadline
      interval = deadline ? deadline - cur_time : WAIT_INTERVAL
      interval = WAIT_INTERVAL if interval > WAIT_INTERVAL
      @lock_cond.wait interval
    end
  end
  true
end

#writable? ⇒ Boolean

Returns true if this writer is still accepting writes. This means it is either running or suspended.

Returns:

  • (Boolean) —

    Returns true if the writer is accepting writes.



295
296
297
298
299
300
 
# File 'lib/google/cloud/logging/async_writer.rb', line 295

def writable?
  ensure_thread
  @lock.synchronize do
    state == :suspended || state == :running
  end
end

#write_entries(entries, log_name: nil, resource: nil, labels: nil) ⇒ Google::Cloud::Logging::AsyncWriter

Asynchronously write one or more log entries to the Stackdriver Logging service.

Unlike the main write_entries method, this method usually does not block. The actual write RPCs will happen in the background, and may be batched with related calls. However, if the queue is full, this method will block until enough space has cleared out.

Examples:

require "google/cloud/logging"

logging = Google::Cloud::Logging.new
async = logging.async_writer

entry = logging.entry payload: "Job started.",
                      log_name: "my_app_log"
entry.resource.type = "gae_app"
entry.resource.labels[:module_id] = "1"
entry.resource.labels[:version_id] = "20150925t173233"

async.write_entries entry

Parameters:

  • entries (Google::Cloud::Logging::Entry, Array<Google::Cloud::Logging::Entry>) —

    One or more entry objects to write. The log entries must have values for all required fields.

  • log_name (String) —

    A default log ID for those log entries in entries that do not specify their own log_name. See also Entry#log_name=.

  • resource (Resource) —

    A default monitored resource for those log entries in entries that do not specify their own resource. See also Entry#resource.

  • labels (Hash{Symbol,String => String}) —

    User-defined key:value items that are added to the labels field of each log entry in entries, except when a log entry specifies its own key:value item with the same key. See also Entry#labels=.

Returns:



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/google/cloud/logging/async_writer.rb', line 145

def write_entries entries, log_name: nil, resource: nil, labels: nil
  ensure_thread
  entries = Array(entries)
  @lock.synchronize do
    fail "AsyncWriter has been stopped" unless writable?
    queue_item = QueueItem.new entries, log_name, resource, labels
    if @queue.empty? || !@queue.last.try_combine(queue_item)
      @queue.push queue_item
    end
    @queue_size += entries.size
    @lock_cond.broadcast
    while @max_queue_size && @queue_size > @max_queue_size
      @lock_cond.wait
    end
  end
  self
end