WIP

include/ipfixprobe/input.hpp

@@ -68,6 +68,12 @@ class InputPlugin : public TelemetryUtils, public Plugin
       std::shared_ptr<telemetry::Directory> plugin_dir, 
       std::shared_ptr<telemetry::Directory> queues_dir);
 
+#ifdef WITH_CTT
+   virtual std::pair<std::string, unsigned> get_ctt_config() const {
+      throw PluginError("CTT is not supported by this input plugin");
+   }
+#endif /* WITH_CTT */
+
 protected:
    virtual void configure_telemetry_dirs(
       std::shared_ptr<telemetry::Directory> plugin_dir, 

input/ndp.cpp

@@ -97,13 +97,31 @@ void NdpPacketReader::init(const char *params)
       m_ctt_metadata = true;
    }
    init_ifc(parser.m_dev);
+   m_device = parser.m_dev;
 }
 
 void NdpPacketReader::close()
 {
    ndpReader.close();
 }
 
+#ifdef WITH_CTT
+
+std::pair<std::string, unsigned> NdpPacketReader::get_ctt_config() const 
+{
+   std::string dev = m_device;
+   int channel_id = 0;
+   std::size_t delimiter_found = m_device.find_last_of(":");
+   if (delimiter_found != std::string::npos) {
+      std::string channel_str = m_device.substr(delimiter_found + 1);
+      dev = m_device.substr(0, delimiter_found);
+      channel_id = std::stoi(channel_str);
+   }
+   return std::make_pair(dev, channel_id);
+}
+
+#endif /* WITH_CTT */
+
 void NdpPacketReader::init_ifc(const std::string &dev)
 {
    if (ndpReader.init_interface(dev) != 0) {

input/ndp.hpp

@@ -73,6 +73,10 @@ class NdpPacketReader : public InputPlugin
       std::shared_ptr<telemetry::Directory> plugin_dir, 
       std::shared_ptr<telemetry::Directory> queues_dir) override;
 
+#ifdef WITH_CTT
+   virtual std::pair<std::string, unsigned> get_ctt_config() const override;
+#endif /* WITH_CTT */
+
 private:
    struct RxStats {
         uint64_t receivedPackets;
@@ -87,6 +91,8 @@ class NdpPacketReader : public InputPlugin
 
    bool m_ctt_metadata = false;
 
+   std::string m_device;
+
    void init_ifc(const std::string &dev);
    int parse_ctt_metadata(const ndp_packet *ndp_packet, Metadata_CTT &ctt);
 };

ipfixprobe.cpp

@@ -358,6 +358,10 @@ bool process_plugin_args(ipxp_conf_t &conf, IpfixprobeOptParser &parser)
          if (storage_plugin == nullptr) {
             throw IPXPError("invalid storage plugin " + storage_name);
          }
+#ifdef WITH_CTT
+         const auto& [device, comp_idx] = input_plugin->get_ctt_config();
+         storage_plugin->set_ctt_config(device, comp_idx);
+#endif /* WITH_CTT */
          storage_plugin->set_queue(output_queue);
          storage_plugin->init(storage_params.c_str());
          storage_plugin->set_telemetry_dir(pipeline_queue_dir);