[WIP] Added logs with level INFO with string F_C_CH for each log, for the f…

confd/templates/nbworker-topology/nbworker-topology.tmpl

@@ -24,3 +24,5 @@ bolts:
     parallelism: 1
   - id: "flows-operations-bolt"
     parallelism: {{ getv "/kilda_storm_nb_worker_flow_operations_parallelism" }}
+  - id: "history-operations-bolt"
+    parallelism: 3

confd/vars/main.yaml

@@ -194,7 +194,7 @@ kilda_storm_stats_workers_count: 2
 
 # Network
 kilda_storm_network_parallelism: 2
-kilda_storm_network_workers_count: 2
+kilda_storm_network_workers_count: 1
 
 # Reroute
 kilda_storm_reroute_parallelism: 2
@@ -207,7 +207,7 @@ kilda_storm_floodlight_router_workers_count: 2
 # NB worker
 kilda_storm_nb_worker_parallelism: 2
 kilda_storm_nb_worker_flow_operations_parallelism: 2
-kilda_storm_nbworker_workers_count: 2
+kilda_storm_nbworker_workers_count: 1
 
 # Server42 control
 kilda_storm_server42_control_parallelism: 2
@@ -217,7 +217,7 @@ kilda_storm_server42_control_count: 2
 kilda_storm_history_parallelism: 2
 kilda_storm_history_bolt_parallelism: 2
 kilda_storm_history_bolt_num_tasks: 2
-kilda_storm_history_workers_count: 2
+kilda_storm_history_workers_count: 1
 
 # Connected devices
 kilda_storm_connected_devices_parallelism: 2

src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java

@@ -55,7 +55,7 @@ public abstract class AbstractBolt extends BaseRichBolt {
     protected boolean active = false;
 
     @Getter
-    protected final PersistenceManager persistenceManager;
+    protected PersistenceManager persistenceManager;
 
     @Getter(AccessLevel.PROTECTED)
     private transient OutputCollector output;

src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/share/history/service/HistoryService.java

@@ -144,6 +144,8 @@ public void store(PortEventData data) {
      */
     public List<FlowEvent> listFlowEvents(String flowId, Instant timeFrom, Instant timeTo, int maxCount) {
         List<FlowEvent> result = new ArrayList<>();
+        log.info("CHUPIN HistoryService, implementation in transactionManager: {}",
+                transactionManager.getImplementation());
         transactionManager.doInTransaction(() -> flowEventRepository
                 .findByFlowIdAndTimeFrame(flowId, timeFrom, timeTo, maxCount)
                 .forEach(entry -> {

src-java/history-topology/history-storm-topology/src/main/java/org/openkilda/wfm/topology/history/bolts/HistoryBolt.java

@@ -27,20 +27,49 @@
 import org.openkilda.wfm.share.zk.ZkStreams;
 import org.openkilda.wfm.share.zk.ZooKeeperBolt;
 
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
 import org.apache.storm.topology.OutputFieldsDeclarer;
 import org.apache.storm.tuple.Fields;
 import org.apache.storm.tuple.Tuple;
 
+import java.util.Map;
+
 public class HistoryBolt extends AbstractBolt {
     private transient HistoryService historyService;
+    private static volatile PersistenceManager PERSISTENCE_MANAGER_INSTANCE;
 
     public HistoryBolt(PersistenceManager persistenceManager, String lifeCycleEventSourceComponent) {
         super(persistenceManager, lifeCycleEventSourceComponent);
     }
 
+
+    /**
+     * Called when a task for this component is initialized within a worker on the cluster.
+     * It provides the bolt with the environment in which the bolt executes.
+     * Static instance of PersistenceManager is initialized at this step.
+     *
+     * @param stormConf The Storm configuration for this bolt.
+     * @param context   This object can be used to get information about this task's place within the topology.
+     * @param collector The collector is used to emit tuples from this bolt.
+     */
+    @Override
+    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+        if (PERSISTENCE_MANAGER_INSTANCE == null) {
+            synchronized (HistoryBolt.class) {
+                if (PERSISTENCE_MANAGER_INSTANCE == null) {
+                    PERSISTENCE_MANAGER_INSTANCE = persistenceManager;
+                    PERSISTENCE_MANAGER_INSTANCE.install();
+                }
+            }
+        }
+        persistenceManager = null;
+        super.prepare(stormConf, context, collector);
+    }
+
     @Override
     protected void init() {
-        historyService = new HistoryService(persistenceManager);
+        historyService = new HistoryService(PERSISTENCE_MANAGER_INSTANCE);
     }
 
     @Override

src-java/kilda-persistence-api/src/main/java/org/openkilda/persistence/PersistenceManager.java

@@ -139,6 +139,20 @@ private PersistenceImplementation getImplementation(PersistenceImplementationTyp
         return implementation;
     }
 
+    /**
+     * Get retries limit.
+     */
+    public int getTransactionRetriesLimit() {
+        return persistenceConfig.getTransactionRetriesLimit();
+    }
+
+    /**
+     * Get retries max delay.
+     */
+    public int getTransactionRetriesMaxDelay() {
+        return persistenceConfig.getTransactionRetriesMaxDelay();
+    }
+
     public void install() {
         PersistenceContextManager.install(this);
     }

src-java/kilda-persistence-api/src/main/java/org/openkilda/persistence/repositories/history/FlowEventRepository.java

@@ -21,13 +21,10 @@
 
 import java.time.Instant;
 import java.util.List;
-import java.util.Optional;
 
 public interface FlowEventRepository extends Repository<FlowEvent> {
     boolean existsByTaskId(String taskId);
 
-    Optional<FlowEvent> findByTaskId(String taskId);
-
     List<FlowEvent> findByFlowIdAndTimeFrame(String flowId, Instant timeFrom, Instant timeTo, int maxCount);
 
     List<FlowStatusView> findFlowStatusesByFlowIdAndTimeFrame(String flowId, Instant timeFrom,

src-java/kilda-persistence-api/src/main/java/org/openkilda/persistence/tx/TransactionManager.java

@@ -22,6 +22,7 @@
 import org.openkilda.persistence.exceptions.PersistenceException;
 import org.openkilda.persistence.exceptions.RecoverablePersistenceException;
 
+import lombok.Getter;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import net.jodah.failsafe.Failsafe;
@@ -37,6 +38,7 @@
  */
 @Slf4j
 public class TransactionManager implements Serializable {
+    @Getter
     private final PersistenceImplementation implementation;
 
     private final int transactionRetriesLimit;

src-java/kilda-persistence-hibernate/src/main/java/org/openkilda/persistence/hibernate/HibernateSessionFactorySupplier.java

@@ -34,7 +34,7 @@
 import org.hibernate.boot.registry.StandardServiceRegistryBuilder;
 import org.hibernate.cfg.AvailableSettings;
 import org.hibernate.context.internal.ManagedSessionContext;
-import org.hibernate.dialect.MySQLDialect;
+import org.hibernate.dialect.MySQL8Dialect;
 
 import java.io.Serializable;
 import java.util.function.Supplier;
@@ -62,14 +62,16 @@ public SessionFactory get() {
     }
 
     private SessionFactory makeHibernateSessionFactory() {
+        log.info("HibernateSessionFactorySupplier makeHibernateSessionFactory");
         StandardServiceRegistry standardRegistry = new StandardServiceRegistryBuilder()
                 .applySetting(AvailableSettings.DRIVER, hibernateConfig.getDriverClass())
                 .applySetting(AvailableSettings.USER, hibernateConfig.getUser())
                 .applySetting(AvailableSettings.PASS, hibernateConfig.getPassword())
                 .applySetting(AvailableSettings.URL, hibernateConfig.getUrl())
-                .applySetting(AvailableSettings.DIALECT, MySQLDialect.class.getName())
+                .applySetting(AvailableSettings.DIALECT, MySQL8Dialect.class.getName())
                 .applySetting(AvailableSettings.CURRENT_SESSION_CONTEXT_CLASS, ManagedSessionContext.class.getName())
                 .applySetting(AvailableSettings.C3P0_IDLE_TEST_PERIOD, 600)  // seconds?
+                .applySetting(AvailableSettings.C3P0_MAX_SIZE, 30)
                 .applySetting(AvailableSettings.C3P0_CONFIG_PREFIX + ".testConnectionOnCheckout", true)
                 .applySetting(AvailableSettings.C3P0_CONFIG_PREFIX + ".preferredTestQuery", "SELECT 1")
                 // TODO(surabujin): detect debugging mode and enable for it

src-java/kilda-persistence-hibernate/src/main/java/org/openkilda/persistence/hibernate/repositories/HibernateGenericRepository.java

@@ -25,9 +25,11 @@
 import org.openkilda.persistence.repositories.Repository;
 import org.openkilda.persistence.tx.TransactionManager;
 
+import lombok.extern.slf4j.Slf4j;
 import org.hibernate.Session;
 import org.hibernate.SessionFactory;
 
+@Slf4j
 public abstract class HibernateGenericRepository<M extends CompositeDataEntity<V>, V, H extends V>
         implements Repository<M> {
     protected final HibernatePersistenceImplementation implementation;
@@ -42,7 +44,7 @@ public void add(M model) {
         if (view instanceof EntityBase) {
             throw new IllegalArgumentException("Entity of class " + model + " already persisted");
         }
-        getTransactionManager().doInTransaction(() -> {
+        getTransactionManagerWithLocalPersistenceImplementation().doInTransaction(() -> {
             H entity = makeEntity(view);
             getSession().persist(entity);
             model.setData(entity);
@@ -63,7 +65,8 @@ public void remove(M model) {
         H hibernateView = (H) view;
         V detachedView = doDetach(model, hibernateView);
 
-        getTransactionManager().doInTransaction(() -> getSession().remove(hibernateView));
+        getTransactionManagerWithLocalPersistenceImplementation()
+                .doInTransaction(() -> getSession().remove(hibernateView));
         model.setData(detachedView);
     }
 
@@ -90,6 +93,12 @@ protected TransactionManager getTransactionManager() {
         return manager.getTransactionManager(implementation.getType());
     }
 
+    protected TransactionManager getTransactionManagerWithLocalPersistenceImplementation() {
+        PersistenceManager pm = PersistenceContextManager.INSTANCE.getPersistenceManager();
+        return new TransactionManager(implementation, pm.getTransactionRetriesLimit(),
+                pm.getTransactionRetriesMaxDelay());
+    }
+
     protected HibernateContextExtension getContextExtension() {
         PersistenceContext context = PersistenceContextManager.INSTANCE.getContextCreateIfMissing();
         return implementation.getContextExtension(context);

src-java/kilda-persistence-hibernate/src/main/java/org/openkilda/persistence/hibernate/repositories/HibernateHistoryFlowEventRepository.java

@@ -31,6 +31,9 @@
 import org.openkilda.persistence.hibernate.entities.history.HibernateFlowEvent_;
 import org.openkilda.persistence.hibernate.utils.UniqueKeyUtil;
 import org.openkilda.persistence.repositories.history.FlowEventRepository;
+import org.openkilda.persistence.tx.TransactionManager;
+
+import lombok.extern.slf4j.Slf4j;
 
 import java.time.Instant;
 import java.util.ArrayList;
@@ -43,6 +46,7 @@
 import javax.persistence.criteria.Predicate;
 import javax.persistence.criteria.Root;
 
+@Slf4j
 public class HibernateHistoryFlowEventRepository
         extends HibernateGenericRepository<FlowEvent, FlowEventData, HibernateFlowEvent>
         implements FlowEventRepository {
@@ -52,18 +56,29 @@ public HibernateHistoryFlowEventRepository(HibernatePersistenceImplementation im
 
     @Override
     public boolean existsByTaskId(String taskId) {
-        return getTransactionManager().doInTransaction(() -> findEntityByTaskId(taskId).isPresent());
-    }
-
-    @Override
-    public Optional<FlowEvent> findByTaskId(String taskId) {
-        return getTransactionManager().doInTransaction(() -> findEntityByTaskId(taskId).map(FlowEvent::new));
+        return getTransactionManagerWithLocalPersistenceImplementation()
+                .doInTransaction(() -> findEntityByTaskId(taskId).isPresent());
     }
 
+    /**
+     * Retrieves a list of {@link FlowEvent} objects filtered by the given flow ID and time frame.
+     * The method performs a transactional operation to fetch the events, maps them to new {@link FlowEvent} objects,
+     * and then reverses the order of the list before returning it.
+     *
+     * @param flowId   The ID of the flow for which events are being retrieved. Cannot be null.
+     * @param timeFrom The start of the time frame for the events.
+     * @param timeTo   The end of the time frame for the events.
+     * @param maxCount The maximum number of events to retrieve.
+     * @return A list of {@link FlowEvent} objects matching the given criteria, in reverse chronological order.
+     *     If no events match the criteria, an empty list is returned.
+     */
     @Override
     public List<FlowEvent> findByFlowIdAndTimeFrame(
             String flowId, Instant timeFrom, Instant timeTo, int maxCount) {
-        List<FlowEvent> results = getTransactionManager().doInTransaction(
+        TransactionManager transactionManager = getTransactionManagerWithLocalPersistenceImplementation();
+        log.info("CHUPIN HibernateHistoryFlowEventRepository findByFlowIdAndTimeFrame, implementation: {}",
+                implementation);
+        List<FlowEvent> results = transactionManager.doInTransaction(
                 () -> fetch(flowId, timeFrom, timeTo, maxCount).stream()
                 .map(FlowEvent::new)
                 .collect(Collectors.toList()));
@@ -74,7 +89,7 @@ public List<FlowEvent> findByFlowIdAndTimeFrame(
     @Override
     public List<FlowStatusView> findFlowStatusesByFlowIdAndTimeFrame(
             String flowId, Instant timeFrom, Instant timeTo, int maxCount) {
-        List<FlowStatusView> results = getTransactionManager().doInTransaction(
+        List<FlowStatusView> results = getTransactionManagerWithLocalPersistenceImplementation().doInTransaction(
                 () -> fetch(flowId, timeFrom, timeTo, maxCount).stream()
                 .flatMap(entry -> entry.getActions().stream())
                 .map(this::extractStatusUpdates)

src-java/kilda-persistence-hibernate/src/main/java/org/openkilda/persistence/hibernate/repositories/HibernateHistoryHaFlowEventRepository.java

@@ -58,7 +58,7 @@ public Optional<HibernateHaFlowEvent> findEntityByTaskId(String taskId) {
     @Override
     public List<HaFlowEvent> findByHaFlowIdAndTimeFrame(String haFlowId,
                                                         Instant timeFrom, Instant timeTo, int maxCount) {
-        List<HaFlowEvent> results = getTransactionManager().doInTransaction(
+        List<HaFlowEvent> results = getTransactionManagerWithLocalPersistenceImplementation().doInTransaction(
                 () -> fetch(haFlowId, timeFrom, timeTo, maxCount).stream()
                         .map(HaFlowEvent::new)
                         .collect(Collectors.toList()));

src-java/kilda-persistence-hibernate/src/main/java/org/openkilda/persistence/hibernate/repositories/HibernateHistoryPortEventRepository.java

@@ -42,7 +42,7 @@ public HibernateHistoryPortEventRepository(HibernatePersistenceImplementation im
     @Override
     public List<PortEvent> findBySwitchIdAndPortNumber(
             SwitchId switchId, int portNumber, Instant start, Instant end) {
-        return getTransactionManager().doInTransaction(
+        return getTransactionManagerWithLocalPersistenceImplementation().doInTransaction(
                 () -> findEntityBySwitchIdAndPortNumber(switchId, portNumber, start, end).stream()
                         .map(PortEvent::new)
                         .collect(Collectors.toList()));

src-java/kilda-persistence-tinkerpop/src/main/java/org/openkilda/persistence/ferma/repositories/FermaFlowEventRepository.java

@@ -35,7 +35,6 @@
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
@@ -59,16 +58,6 @@ public boolean existsByTaskId(String taskId) {
         }
     }
 
-    @Override
-    public Optional<FlowEvent> findByTaskId(String taskId) {
-        List<? extends FlowEventFrame> flowEventFrames = framedGraph().traverse(g -> g.V()
-                .hasLabel(FlowEventFrame.FRAME_LABEL)
-                .has(FlowEventFrame.TASK_ID_PROPERTY, taskId))
-                .toListExplicit(FlowEventFrame.class);
-        return flowEventFrames.isEmpty() ? Optional.empty() : Optional.of(flowEventFrames.get(0))
-                .map(FlowEvent::new);
-    }
-
     @Override
     public List<FlowEvent> findByFlowIdAndTimeFrame(String flowId, Instant timeFrom, Instant timeTo, int maxCount) {
         return framedGraph().traverse(g -> {

src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/bolts/HistoryOperationsBolt.java

@@ -57,8 +57,7 @@ public HistoryOperationsBolt(PersistenceManager persistenceManager) {
     @Override
     public void init() {
         super.init();
-
-        historyService = new HistoryService(persistenceManager);
+        historyService = new HistoryService(PERSISTENCE_MANAGER_INSTANCE);
     }
 
     @Override