Class: Fluent::Plugin::GcloudPubSubInput

Inherits:
Input
  • Object
show all
Defined in:
lib/fluent/plugin/in_gcloud_pubsub.rb

Defined Under Namespace

Classes: FailedParseError, RPCServlet

Constant Summary collapse

DEFAULT_PARSER_TYPE = 
"json"

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object

rubocop:disable Metrics/MethodLength



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 113

def configure(conf)
  compat_parameters_convert(conf, :parser)
  super
  @rpc_srv = nil
  @rpc_thread = nil
  @stop_pull = false

  @extract_tag = if @tag_key.nil?
                   method(:static_tag)
                 else
                   method(:dynamic_tag)
                 end

  @parser = parser_create

  @messages_pulled =
    Fluent::GcloudPubSub::Metrics.register_or_existing(:"#{@metric_prefix}_messages_pulled") do
      ::Prometheus::Client.registry.histogram(
        :"#{@metric_prefix}_messages_pulled",
        "Number of Pub/Sub messages pulled by the subscriber on each invocation",
        {},
        [0, 1, 10, 50, 100, 250, 500, 1000],
      )
    end

  @messages_pulled_bytes =
    Fluent::GcloudPubSub::Metrics.register_or_existing(:"#{@metric_prefix}_messages_pulled_bytes") do
      ::Prometheus::Client.registry.histogram(
        :"#{@metric_prefix}_messages_pulled_bytes",
        "Total size in bytes of the Pub/Sub messages pulled by the subscriber on each invocation",
        {},
        [100, 1000, 10_000, 100_000, 1_000_000, 5_000_000, 10_000_000],
      )
    end

  @pull_errors =
    Fluent::GcloudPubSub::Metrics.register_or_existing(:"#{@metric_prefix}_pull_errors_total") do
      ::Prometheus::Client.registry.counter(
        :"#{@metric_prefix}_pull_errors_total",
        "Errors encountered while pulling or processing messages",
        {},
      )
    end
end

#shutdown ⇒ Object 



174
175
176
177
178
179
180
181
182
183
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 174

def shutdown
  if @rpc_srv
    @rpc_srv.shutdown
    @rpc_srv = nil
  end
  @rpc_thread = nil if @rpc_thread
  @stop_subscribing = true
  @subscribe_threads.each(&:join)
  super
end

#start ⇒ Object

rubocop:enable Metrics/MethodLength



159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 159

def start
  super
  start_rpc if @enable_rpc

  @subscriber = Fluent::GcloudPubSub::Subscriber.new @project, @key, @topic, @subscription
  log.debug "connected subscription:#{@subscription} in project #{@project}"

  @emit_guard = Mutex.new
  @stop_subscribing = false
  @subscribe_threads = []
  @pull_threads.times do |idx|
    @subscribe_threads.push thread_create("in_gcloud_pubsub_subscribe_#{idx}".to_sym, &method(:subscribe))
  end
end

#start_pull ⇒ Object 



190
191
192
193
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 190

def start_pull
  @stop_pull = false
  log.info "start pull from subscription:#{@subscription}"
end

#status_of_pull ⇒ Object 



195
196
197
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 195

def status_of_pull
  @stop_pull ? "stopped" : "started"
end

#stop_pull ⇒ Object 



185
186
187
188
 
# File 'lib/fluent/plugin/in_gcloud_pubsub.rb', line 185

def stop_pull
  @stop_pull = true
  log.info "stop pull from subscription:#{@subscription}"
end