Class: OpenC3::InterfaceMicroservice

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

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
  @interface.tlm_target_names.each do |target_name|
    # Initialize the target's packet counters based on the Topic stream
    # Prevents packet count resetting to 0 when interface restarts
    begin
      System.telemetry.packets(target_name).each do |packet_name, packet|
        topic = "#{@scope}__TELEMETRY__{#{target_name}}__#{packet_name}"
        msg_id, msg_hash = Topic.get_newest_message(topic)
        if msg_id
          packet.received_count = msg_hash['received_count'].to_i
        else
          packet.received_count = 0
        end
      end
    rescue
      # Handle targets without telemetry
    end
  end
  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
  @sync_packet_count_data = {}
  @sync_packet_count_time = nil
  @sync_packet_count_delay_seconds = 1.0 # Sync packet counts every second
  @interface.options.each do |option_name, option_values|
    if 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'
      @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, scope: @scope)
  else
    @handler_thread = RouterTlmHandlerThread.new(@interface, self, logger: @logger, metric: @metric, scope: @scope)
  end
  @handler_thread.start
end

Instance Method Details

#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.

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

def attempting(*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

#connect ⇒ `Object`

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

def connect
  @logger.info "#{@interface.name}: Connect #{@interface.connection_string}"
  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 849

def disconnect(allow_reconnect = true)
  return if @interface.state == 'DISCONNECTED' && !@interface.connected?

  # Synchronize the calls to @interface.disconnect since it takes an unknown
  # amount of time. If two calls to disconnect stack up, the if statement
  # should avoid multiple calls to disconnect.
  @mutex.synchronize do
    begin
      @interface.disconnect if @interface.connected?
    rescue => e
      @logger.error "Disconnect: #{@interface.name}: #{e.formatted}"
    end
  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.
  if allow_reconnect and @interface.auto_reconnect and @interface.state != 'DISCONNECTED'
    attempting()
    if !@cancel_thread
      # @logger.debug "reconnect delay: #{@interface.reconnect_delay}"
      @interface_thread_sleeper.sleep(@interface.reconnect_delay)
    end
  else
    @interface.state = 'DISCONNECTED'
    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

#graceful_kill ⇒ `Object`



908
909
910

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

def graceful_kill
  # Just to avoid warning
end

#handle_connection_failed(connection, connect_error) ⇒ `Object`

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

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)
        OpenC3.write_exception_file(connect_error)
        @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 804

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)
        OpenC3.write_exception_file(err)
        @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 724

def handle_packet(packet)
  InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope)
  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]
    sync_tlm_packet_counts(packet)
    TelemetryTopic.write_packet(packet, queued: @queued, scope: @scope)
  end
end

#run ⇒ `Object`

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

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
  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}: Stopped packet reading"
end

#shutdown(_sig = nil) ⇒ `Object`

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

def shutdown(_sig = nil)
  @logger.info "#{@interface ? @interface.name : @name}: shutdown requested"
  stop()
  super()
end

#stop ⇒ `Object`

Disconnect from the interface and stop the thread

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

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

#sync_tlm_packet_counts(packet) ⇒ `Object`

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

def sync_tlm_packet_counts(packet)
  if @sync_packet_count_delay_seconds <= 0 or $openc3_redis_cluster
    # Perfect but slow method
    packet.received_count = TargetModel.increment_telemetry_count(packet.target_name, packet.packet_name, 1, scope: @scope)
  else
    # Eventually consistent method
    # Only sync every period (default 1 second) to avoid hammering Redis
    # This is a trade off between speed and accuracy
    # The packet count is eventually consistent
    @sync_packet_count_data[packet.target_name] ||= {}
    @sync_packet_count_data[packet.target_name][packet.packet_name] ||= 0
    @sync_packet_count_data[packet.target_name][packet.packet_name] += 1

    # Ensures counters change between syncs
    packet.received_count += 1

    # Check if we need to sync the packet counts
    if @sync_packet_count_time.nil? or (Time.now - @sync_packet_count_time) > @sync_packet_count_delay_seconds
      @sync_packet_count_time = Time.now

      inc_count = 0
      # Use pipeline to make this one transaction
      result = Store.redis_pool.pipelined do
        # Increment global counters for packets received
        @sync_packet_count_data.each do |target_name, packet_data|
          packet_data.each do |packet_name, count|
            TargetModel.increment_telemetry_count(target_name, packet_name, count, scope: @scope)
            inc_count += 1
          end
        end
        @sync_packet_count_data = {}

        # Get all the packet counts with the global counters
        @interface.tlm_target_names.each do |target_name|
          TargetModel.get_all_telemetry_counts(target_name, scope: @scope)
        end
        TargetModel.get_all_telemetry_counts('UNKNOWN', scope: @scope)
      end
      @interface.tlm_target_names.each do |target_name|
        result[inc_count].each do |packet_name, count|
          update_packet = System.telemetry.packet(target_name, packet_name)
          update_packet.received_count = count.to_i
        end
        inc_count += 1
      end
      result[inc_count].each do |packet_name, count|
        update_packet = System.telemetry.packet('UNKNOWN', packet_name)
        update_packet.received_count = count.to_i
      end
    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

#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

#sync_tlm_packet_counts(packet) ⇒ Object

#initialize(name) ⇒ `InterfaceMicroservice`