Class: OpenC3::InterfaceMicroservice

# File 'lib/openc3/microservices/interface_microservice.rb', line 559

def initialize(name)
  @mutex = Mutex.new
  super(name)

  @interface_or_router = self.class.name.to_s.split("Microservice")[0].upcase.split("::")[-1]
  if @interface_or_router == 'INTERFACE'
    @metric.set(name: 'interface_tlm_total', value: @count, type: 'counter')
  else
    @metric.set(name: 'router_cmd_total', value: @count, type: 'counter')
  end

  interface_name = name.split("__")[2]
  if @interface_or_router == 'INTERFACE'
    @interface = InterfaceModel.get_model(name: interface_name, scope: @scope).build
  else
    @interface = RouterModel.get_model(name: interface_name, scope: @scope).build
  end
  @interface.name = interface_name
  # Map the interface to the interface's targets
  @interface.target_names.each do |target_name|
    target = System.targets[target_name]
    target.interface = @interface
  end
  TargetModel.init_tlm_packet_counts(@interface.tlm_target_names, scope: @scope)
  if @interface.connect_on_startup
    @interface.state = 'ATTEMPTING'
  else
    @interface.state = 'DISCONNECTED'
  end
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json(), scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json(), scope: @scope)
  end

  @queued = false
  @interface.options.each do |option_name, option_values|
    # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting
    if option_name.upcase == 'UPDATE_INTERVAL' or option_name.upcase == 'OPTIMIZE_THROUGHPUT'
      @queued = true
      update_interval = option_values[0].to_f
      EphemeralStoreQueued.instance.set_update_interval(update_interval)
      StoreQueued.instance.set_update_interval(update_interval)
    end
    if option_name.upcase == 'SYNC_PACKET_COUNT_DELAY_SECONDS'
      TargetModel.sync_packet_count_delay_seconds = option_values[0].to_f
    end
  end

  @interface_thread_sleeper = Sleeper.new
  @cancel_thread = false
  @connection_failed_messages = []
  @connection_lost_messages = []
  if @interface_or_router == 'INTERFACE'
    @handler_thread = InterfaceCmdHandlerThread.new(@interface, self, logger: @logger, metric: @metric, db_shard: @db_shard, scope: @scope)
  else
    @handler_thread = RouterTlmHandlerThread.new(@interface, self, logger: @logger, metric: @metric, db_shard: @db_shard, scope: @scope)
  end
  @handler_thread.start
end

Instance Method Details

#attempt_connection(*params) ⇒ `Object`

Sets the state to 'ATTEMPTING', first rebuilding the interface/router if parameters are given, so the run method performs the actual connection. Unlike attempting() this always transitions. The reconnect path in disconnect() requires that, since @interface.disconnect may have raised or left connected? true, which would make attempting() ignore the request and leave the interface stuck in 'CONNECTED' with no way back to a connection.

# File 'lib/openc3/microservices/interface_microservice.rb', line 642

def attempt_connection(*params)
  unless params.empty?
    @interface.disconnect()
    # Build New Interface, this can fail if passed bad parameters
    new_interface = @interface.class.new(*params)
    @interface.copy_to(new_interface)

    # Replace interface for targets
    @interface.target_names.each do |target_name|
      target = System.targets[target_name]
      target.interface = new_interface
    end
    @interface = new_interface

    # Update the model
    if @interface_or_router == 'INTERFACE'
      interface_model = InterfaceModel.get(name: @interface.name, scope: @scope)
      # config_params[0] is the filename so set the rest
      interface_model['config_params'][1..-1] = *params
      InterfaceModel.set(interface_model, scope: @scope)
    else
      router_model = RouterModel.get(name: @interface.name, scope: @scope)
      # config_params[0] is the filename so set the rest
      router_model['config_params'][1..-1] = *params
      RouterModel.set(router_model, scope: @scope)
    end
  end

  @interface.state = 'ATTEMPTING'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
  end
  @interface # Return the interface/router since we may have recreated it
# Need to rescue Exception so we cover LoadError
rescue Exception => e
  @logger.error("Attempting connection #{@interface.connection_string} failed due to #{e.message}")
  if SignalException === e
    @logger.info "#{@interface.name}: Closing from signal"
    @cancel_thread = true
  end
  @interface # Return the original interface/router in case of error
end

#attempting(*params) ⇒ `Object`

Called to connect the interface/router. It takes optional parameters to rebuilt the interface/router. Once we set the state to 'ATTEMPTING' the run method handles the actual connection. Connecting an interface/router which is already CONNECTED is a no-op. Without this the existing (working) connection would be torn down and rebuilt which can take up to the read_timeout to detect. Callers who want to force a reconnect should disconnect first or pass new parameters.

# File 'lib/openc3/microservices/interface_microservice.rb', line 627

def attempting(*params)
  if params.empty? and @interface.state == 'CONNECTED' and @interface.connected?
    @logger.info "#{@interface.name}: Connect ignored, already connected"
    return @interface
  end

  attempt_connection(*params)
end

#connect ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 858

def connect
  @logger.info "#{@interface.name}: Connect #{@interface.connection_string}"
  # Interface connect implementations typically overwrite their stream / socket
  # so cleanly close any existing connection rather than leaking it
  begin
    @interface.disconnect if @interface.connected?
  rescue => e
    @logger.error "Disconnect: #{@interface.name}: #{e.formatted}"
  end
  begin
    @interface.connect
    @interface.post_connect
  rescue Exception => e
    begin
      @interface.disconnect # Ensure disconnect is called at least once on a partial connect
    rescue Exception
      # We want to report any connect errors, not disconnect in this case
    end
    raise e
  end
  @interface.state = 'CONNECTED'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
  end
  @logger.info "#{@interface.name}: Connection Success"
end

#disconnect(allow_reconnect = true) ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 887

def disconnect(allow_reconnect = true)
  reconnect = false

  # Two threads reach here for a single connection loss: the cmd handler
  # thread servicing a disconnect directive, and the run thread coming back
  # out of read (or out of the connection maintenance sleep). The redundant
  # check below and the state change that records the disconnect must be in
  # the same critical section, otherwise the second thread reads the state
  # before the first has updated it and disconnects the interface twice.
  @mutex.synchronize do
    # A disconnect has already been performed so there is nothing left to do
    return if @interface.state == 'DISCONNECTED' && !@interface.connected?

    # Call disconnect without consulting connected? so any resources the
    # interface is still holding are cleaned up. It takes an unknown amount
    # of time which is the other reason for the mutex.
    begin
      @interface.disconnect
    rescue => e
      @logger.error "Disconnect: #{@interface.name}: #{e.formatted}"
    end

    # If the interface is set to auto_reconnect then delay so the thread
    # can come back around and allow the interface a chance to reconnect.
    # Skip reconnect if stop() has been called to avoid re-creating the status model
    reconnect = allow_reconnect && @interface.auto_reconnect && @interface.state != 'DISCONNECTED' && !@cancel_thread
    if reconnect
      attempt_connection()
    else
      @interface.state = 'DISCONNECTED'
      unless @cancel_thread
        if @interface_or_router == 'INTERFACE'
          InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
        else
          RouterStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
        end
      end
    end
  end

  # Sleep outside the mutex so stop() and connect() are not blocked for the
  # whole reconnect delay
  # @logger.debug "reconnect delay: #{@interface.reconnect_delay}"
  @interface_thread_sleeper.sleep(@interface.reconnect_delay) if reconnect && !@cancel_thread
end

#graceful_kill ⇒ `Object`



968
969
970

# File 'lib/openc3/microservices/interface_microservice.rb', line 968

def graceful_kill
  # Just to avoid warning
end

#handle_connection_failed(connection, connect_error) ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 814

def handle_connection_failed(connection, connect_error)
  @error = connect_error
  @logger.error "#{@interface.name}: Connection #{connection} failed due to #{connect_error.formatted(false, false)}"
  case connect_error
  when SignalException
    @logger.info "#{@interface.name}: Closing from signal"
    @cancel_thread = true
  when Errno::ECONNREFUSED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::ENOTSOCK, Errno::EHOSTUNREACH, IOError
    # Do not write an exception file for these extremely common cases
  else
    if RuntimeError === connect_error and (connect_error.message =~ /canceled/ or connect_error.message =~ /timeout/)
      # Do not write an exception file for these extremely common cases
    else
      @logger.error "#{@interface.name}: #{connect_error.formatted}"
      unless @connection_failed_messages.include?(connect_error.message)
        @connection_failed_messages << connect_error.message
      end
    end
  end
  disconnect() # Ensure we do a clean disconnect
end

#handle_connection_lost(err = nil, reconnect: true) ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 836

def handle_connection_lost(err = nil, reconnect: true)
  if err
    @error = err
    @logger.info "#{@interface.name}: Connection Lost: #{err.formatted(false, false)}"
    case err
    when SignalException
      @logger.info "#{@interface.name}: Closing from signal"
      @cancel_thread = true
    when Errno::ECONNABORTED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::EBADF, Errno::ENOTSOCK, IOError
      # Do not write an exception file for these extremely common cases
    else
      @logger.error "#{@interface.name}: #{err.formatted}"
      unless @connection_lost_messages.include?(err.message)
        @connection_lost_messages << err.message
      end
    end
  else
    @logger.info "#{@interface.name}: Connection Lost"
  end
  disconnect(reconnect) # Ensure we do a clean disconnect
end

#handle_packet(packet) ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 756

def handle_packet(packet)
  # Skip status update if stop() has been called to avoid re-creating the status model
  InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope) unless @cancel_thread
  packet.received_time = Time.now.sys unless packet.received_time

  if packet.stored
    # Stored telemetry does not update the current value table
    identified_packet = System.telemetry.identify_and_define_packet(packet, @interface.tlm_target_names)
  else
    # Identify and update packet
    if packet.identified?
      begin
        # Preidentifed packet - place it into the current value table
        identified_packet = System.telemetry.update!(packet.target_name,
                                                     packet.packet_name,
                                                     packet.buffer)
      rescue RuntimeError
        # Packet identified but we don't know about it
        # Clear packet_name and target_name and try to identify
        @logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}"
        packet.target_name = nil
        packet.packet_name = nil
        identified_packet = System.telemetry.identify!(packet.buffer,
                                                       @interface.tlm_target_names)
      end
    else
      # Packet needs to be identified
      identified_packet = System.telemetry.identify!(packet.buffer,
                                                     @interface.tlm_target_names)
    end
  end

  if identified_packet
    identified_packet.received_time = packet.received_time
    identified_packet.stored = packet.stored
    identified_packet.extra = packet.extra
    packet = identified_packet
  else
    unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer)
    unknown_packet.received_time = packet.received_time
    unknown_packet.stored = packet.stored
    unknown_packet.extra = packet.extra
    packet = unknown_packet
    json_hash = CvtModel.build_json_from_packet(packet)
    CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, queued: @queued, scope: @scope)
    num_bytes_to_print = [UNKNOWN_BYTES_TO_PRINT, packet.length].min
    data = packet.buffer(false)[0..(num_bytes_to_print - 1)]
    prefix = data.each_byte.map { | byte | sprintf("%02X", byte) }.join()
    @logger.warn "#{@interface.name} #{packet.target_name} packet length: #{packet.length} starting with: #{prefix}"
  end

  # Write to stream
  if @interface.tlm_target_enabled[packet.target_name]
    TargetModel.sync_tlm_packet_counts(packet, @interface.tlm_target_names, scope: @scope)
    TelemetryTopic.write_packet(packet, queued: @queued, scope: @scope)
  end
end

#run ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 687

def run
  begin
    if @interface.read_allowed?
      @logger.info "#{@interface.name}: Starting packet reading"
    else
      @logger.info "#{@interface.name}: Starting connection maintenance"
    end
    while true
      break if @cancel_thread

      case @interface.state
      when 'DISCONNECTED'
        # Just wait to see if we should connect later
        @interface_thread_sleeper.sleep(1)
      when 'ATTEMPTING'
        begin
          @mutex.synchronize do
            # We need to make sure connect is not called after stop() has been called
            connect() unless @cancel_thread
          end
        rescue Exception => e
          handle_connection_failed(@interface.connection_string, e)
          break if @cancel_thread
        end
      when 'CONNECTED'
        if @interface.read_allowed?
          begin
            packet = @interface.read
            if packet
              handle_packet(packet)
              @count += 1
              if @interface_or_router == 'INTERFACE'
                @metric.set(name: 'interface_tlm_total', value: @count, type: 'counter')
              else
                @metric.set(name: 'router_cmd_total', value: @count, type: 'counter')
              end
            else
              @logger.info "#{@interface.name}: Internal disconnect requested (returned nil)"
              handle_connection_lost()
              break if @cancel_thread
            end
          rescue Exception => e
            handle_connection_lost(e)
            break if @cancel_thread
          end
        else
          @interface_thread_sleeper.sleep(1)
          handle_connection_lost() if !@interface.connected?
        end
      end
    end
  rescue Exception => e
    unless SystemExit === e or SignalException === e
      @logger.error "#{@interface.name}: Packet reading thread died: #{e.formatted}"
      OpenC3.handle_fatal_exception(e)
    end
    # Try to do clean disconnect because we're going down
    disconnect(false)
  end
  unless @cancel_thread
    if @interface_or_router == 'INTERFACE'
      InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
    else
      RouterStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
    end
  end
  @logger.info "#{@interface.name}: Stopped packet reading"
end

#shutdown(_sig = nil) ⇒ `Object`

# File 'lib/openc3/microservices/interface_microservice.rb', line 954

def shutdown(_sig = nil)
  @logger.info "#{@interface ? @interface.name : @name}: shutdown requested"
  stop()
  if @interface and @interface.stream_log_pair
    threads = @interface.stream_log_pair.shutdown
    # Wait for all the logging threads to move files to buckets
    threads.flatten.compact.each do |thread|
      thread.join
    end
    @interface.stream_log_pair.cleanup
  end
  super()
end

#stop ⇒ `Object`

Disconnect from the interface and stop the thread

# File 'lib/openc3/microservices/interface_microservice.rb', line 934

def stop
  @logger.info "#{@interface ? @interface.name : @name}: stop requested"
  @mutex.synchronize do
    # Need to make sure that @cancel_thread is set and the interface disconnected within
    # mutex to ensure that connect() is not called when we want to stop()
    @cancel_thread = true
    @handler_thread.stop if @handler_thread
    @interface_thread_sleeper.cancel if @interface_thread_sleeper
    if @interface
      @interface.disconnect
      if @interface_or_router == 'INTERFACE'
        valid_interface = InterfaceStatusModel.get_model(name: @interface.name, scope: @scope)
      else
        valid_interface = RouterStatusModel.get_model(name: @interface.name, scope: @scope)
      end
      valid_interface.destroy if valid_interface
    end
  end
end

Class: OpenC3::InterfaceMicroservice

Direct Known Subclasses

Constant Summary collapse

Instance Attribute Summary

Attributes inherited from Microservice

Instance Method Summary collapse

Methods inherited from Microservice

Constructor Details

#initialize(name) ⇒ InterfaceMicroservice

Instance Method Details

#attempt_connection(*params) ⇒ Object

#attempting(*params) ⇒ Object

#connect ⇒ Object

#disconnect(allow_reconnect = true) ⇒ Object

#graceful_kill ⇒ Object

#handle_connection_failed(connection, connect_error) ⇒ Object

#handle_connection_lost(err = nil, reconnect: true) ⇒ Object

#handle_packet(packet) ⇒ Object

#run ⇒ Object

#shutdown(_sig = nil) ⇒ Object

#stop ⇒ Object

#initialize(name) ⇒ `InterfaceMicroservice`