ingest.h

#ifndef _INGEST_H
#define _INGEST_H
/*
 * Fledge reading ingest.
 *
 * Copyright (c) 2018 OSisoft, LLC
 *
 * Released under the Apache 2.0 Licence
 *
 * Author: Mark Riddoch, Massimiliano Pinto, Amandeep Singh Arora
 */
#include <storage_client.h>
#include <reading.h>
#include <logger.h>
#include <vector>
#include <queue>
#include <thread>
#include <chrono>
#include <mutex>
#include <sstream>
#include <unordered_set>
#include <condition_variable>
#include <filter_plugin.h>
#include <filter_pipeline.h>
#include <asset_tracking.h>
#include <service_handler.h>
#include <set>
#include <perfmonitors.h>
 
#define SERVICE_NAME  "Fledge South"
 
#define INGEST_SUFFIX   "-Ingest"   // Suffix for per service ingest statistic
 
#define FLUSH_STATS_INTERVAL 5      // Period between flushing of stats to storage (seconds)
 
#define STATS_UPDATE_FAIL_THRESHOLD 10  // After this many update fails try creating new stats
 
#define DEPRECATED_CACHE_AGE    600 // Maximum allowed aged of the deprecated asset cache
 
/*
 * Constants related to flow control for async south services.
 *
 */
#define AFC_SLEEP_INCREMENT 20  // Number of milliseconds to wait for readings to drain
#define AFC_SLEEP_MAX       200 // Maximum sleep tiem in ms between tests
#define AFC_MAX_WAIT        5000    // Maximum amount of time we wait for the queue to drain
 
class IngestRate;
 
// Enum for service buffering type
enum class ServiceBufferingType {
    UNLIMITED,
    LIMITED
};
 
// Enum for discard policy
enum class DiscardPolicy {
    DISCARD_OLDEST,
    REDUCE_FIDELITY,
    DISCARD_NEWEST
};
 
#define SERVICE_BUFFER_BUFFER_TYPE_DEFAULT  ServiceBufferingType::UNLIMITED
#define SERVICE_BUFFER_DISCARD_POLICY_DEFAULT   DiscardPolicy::DISCARD_OLDEST
#define SERVICE_BUFFER_SIZE_DEFAULT     1000
#define SERVICE_BUFFER_SIZE_MIN     1000
 
class Ingest : public ServiceHandler {
 
public:
    Ingest(StorageClient& storage,
        const std::string& serviceName,
        const std::string& pluginName,
        ManagementClient *mgmtClient);
    ~Ingest();
 
    void        ingest(const Reading& reading);
    void        ingest(const std::vector<Reading *> *vec);
    void        start(long timeout, unsigned int threshold);
    bool        running();
        bool        isStopping();
    bool        isRunning() { return !m_shutdown; };
    void        processQueue();
    void        waitForQueue();
    size_t      queueLength();
    void        updateStats(void);
    int         createStatsDbEntry(const std::string& assetName);
 
    bool        loadFilters(const std::string& categoryName);
    static void passToOnwardFilter(OUTPUT_HANDLE *outHandle,
                       READINGSET* readings);
    static void useFilteredData(OUTPUT_HANDLE *outHandle,
                    READINGSET* readings);
 
    void        setTimeout(const long timeout) { m_timeout = timeout; };
    void        setThreshold(const unsigned int threshold) { m_queueSizeThreshold = threshold; };
    void        configChange(const std::string&, const std::string&);
    void        configChildCreate(const std::string& , const std::string&, const std::string&){};
    void        configChildDelete(const std::string& , const std::string&){};
    void        shutdown() {};  // Satisfy ServiceHandler
    void        restart() {};   // Satisfy ServiceHandler
    void        unDeprecateAssetTrackingRecord(AssetTrackingTuple* currentTuple,
                            const std::string& assetName,
                            const std::string& event);
    void        unDeprecateStorageAssetTrackingRecord(StorageAssetTrackingTuple* currentTuple,
                            const std::string& assetName,
                            const std::string&,
                            const unsigned int&);
    void        setStatistics(const std::string& option);
 
    std::string     getStringFromSet(const std::set<std::string> &dpSet);
    void        setFlowControl(unsigned int lowWater, unsigned int highWater) { m_lowWater = lowWater; m_highWater = highWater; };
    void        setResourceLimit(ServiceBufferingType serviceBufferingType, unsigned long serviceBufferSize, DiscardPolicy discardPolicy);
    void        flowControl();
    void        setPerfMon(PerformanceMonitor *mon)
            {
                m_performance = mon;
            };
    void        configureRateMonitor(long interval, long factor);
 
    // Debugger entry points
    bool        attachDebugger()
            {
                if (m_filterPipeline)
                {
                    m_debuggerAttached = true;
                    return m_filterPipeline->attachDebugger();
                }
                return false;
            };
    void        detachDebugger()
            {
                if (m_filterPipeline)
                {
                    m_debuggerAttached = false;
                    m_debuggerBufferSize = 1;
                    m_filterPipeline->detachDebugger();
                }
            };
    void        setDebuggerBuffer(unsigned int size)
            {
                if (m_filterPipeline)
                {
                    m_debuggerBufferSize = size;
                    m_filterPipeline->setDebuggerBuffer(size);
                }
            };
    std::string getDebuggerBuffer()
            {
                std::string rval;
                if (m_filterPipeline)
                    rval = m_filterPipeline->getDebuggerBuffer();
                return rval;
            };
    void        isolate(bool isolate)
            {
                std::lock_guard<std::mutex> guard(m_isolateMutex);
                m_isolate = isolate;
            };
    bool        isolated()
            {
                std::lock_guard<std::mutex> guard(m_isolateMutex);
                return m_isolate;
            };
    void        replayDebugger()
            {
                if (m_filterPipeline)
                    m_filterPipeline->replayDebugger();
            };
 
private:
    void                signalStatsUpdate() {
                        // Signal stats thread to update stats
                        std::lock_guard<std::mutex> guard(m_statsMutex);
                        m_statsCv.notify_all();
                    };
    void                logDiscardedStat() {
                        std::lock_guard<std::mutex> guard(m_statsMutex);
                        m_discardedReadings++;
                    };
    long                calculateWaitTime();
    int                 createServiceStatsDbEntry();
    void                discardOldest();
    void                discardNewest();
    void                reduceFidelity();
    void                enforceResourceLimits();
 
    StorageClient&          m_storage;
    long                m_timeout;
    bool                m_shutdown;
    unsigned int            m_queueSizeThreshold;
    bool                m_running;
    std::string             m_serviceName;
    std::string             m_pluginName;
    ManagementClient        *m_mgtClient;
    // New data: queued
    std::vector<Reading *>*     m_queue;
    std::mutex          m_qMutex;
    std::mutex          m_statsMutex;
    std::mutex          m_pipelineMutex;
    std::thread*            m_thread;
    std::thread*            m_statsThread;
    Logger*             m_logger;
    std::condition_variable     m_cv;
    std::condition_variable     m_statsCv;
    // Data ready to be filtered/sent
    std::vector<Reading *>*     m_data;
    std::vector<std::vector<Reading *>*>
                    m_resendQueues;
    std::queue<std::vector<Reading *>*>
                    m_fullQueues;
    std::mutex          m_fqMutex;
    unsigned int            m_discardedReadings; // discarded readings since last update to statistics table
    FilterPipeline*         m_filterPipeline;
    
    std::unordered_set<std::string> statsDbEntriesCache;  // confirmed stats table entries
    std::map<std::string, int>  statsPendingEntries;  // pending stats table entries
    bool                m_highLatency;        // Flag to indicate we are exceeding latency request
    bool                m_10Latency;          // Latency within 10%
    time_t              m_reportedLatencyTime;// Last tiem we reported high latency
    int             m_failCnt;
    bool                m_storageFailed;
    int             m_storesFailed;
    int             m_statsUpdateFails;
    enum { STATS_BOTH, STATS_ASSET, STATS_SERVICE }
                    m_statisticsOption;
    unsigned int            m_highWater;
    unsigned int            m_lowWater;
    AssetTrackingTable      *m_deprecated;
    time_t              m_deprecatedAgeOut;
    time_t              m_deprecatedAgeOutStorage;
    PerformanceMonitor      *m_performance;
    std::mutex          m_useDataMutex;
    IngestRate          *m_ingestRate;
    std::mutex          m_isolateMutex;
    bool                m_isolate;
    bool                m_debuggerAttached;
    unsigned int            m_debuggerBufferSize;
    std::atomic<ServiceBufferingType>           m_serviceBufferingType;
    std::atomic<unsigned int>           m_serviceBufferSize;
    std::atomic<DiscardPolicy>          m_discardPolicy;
    bool m_resourceGovernorActive{false}; // Tracks if the resource governor is active
    time_t m_lastFidelityReductionTimestamp{0}; // Used for "Reduce Fidelity"
};
 
#endif