path: root/cpp/src/qpid/management/ManagementAgent.h

#ifndef _ManagementAgent_
#define _ManagementAgent_

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
#include "qpid/broker/BrokerImportExport.h"
#include "qpid/Options.h"
#include "qpid/broker/Exchange.h"
#include "qpid/framing/Uuid.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/Timer.h"
#include "qpid/broker/ConnectionToken.h"
#include "qpid/management/ManagementObject.h"
#include "qpid/management/ManagementEvent.h"
#include "qpid/management/Manageable.h"
#include "qmf/org/apache/qpid/broker/Agent.h"
#include "qpid/types/Variant.h"
#include <qpid/framing/AMQFrame.h>
#include <qpid/framing/FieldValue.h>
#include <qpid/framing/ResizableBuffer.h>
#include <memory>
#include <string>
#include <map>

namespace qpid {
namespace broker {
class ConnectionState;
}
namespace management {

class ManagementAgent
{
private:

    int threadPoolSize;

public:
    typedef enum {
    SEV_EMERG = 0,
    SEV_ALERT = 1,
    SEV_CRIT  = 2,
    SEV_ERROR = 3,
    SEV_WARN  = 4,
    SEV_NOTE  = 5,
    SEV_INFO  = 6,
    SEV_DEBUG = 7,
    SEV_DEFAULT = 8
    } severity_t;


    ManagementAgent (const bool qmfV1, const bool qmfV2);
    virtual ~ManagementAgent ();

    /** Called before plugins are initialized */
    void configure       (const std::string& dataDir, uint16_t interval,
                          qpid::broker::Broker* broker, int threadPoolSize);
    /** Called after plugins are initialized. */
    void pluginsInitialized();

    /** Called by cluster to suppress management output during update. */
    void suppress(bool s) { suppressed = s; }

    void setName(const std::string& vendor,
                 const std::string& product,
                 const std::string& instance="");
    void getName(std::string& vendor, std::string& product, std::string& instance);
    const std::string& getAddress();

    void setInterval(uint16_t _interval) { interval = _interval; }
    void setExchange(qpid::broker::Exchange::shared_ptr mgmtExchange,
                     qpid::broker::Exchange::shared_ptr directExchange);
    void setExchangeV2(qpid::broker::Exchange::shared_ptr topicExchange,
                       qpid::broker::Exchange::shared_ptr directExchange);

    int  getMaxThreads   () { return threadPoolSize; }
    QPID_BROKER_EXTERN void registerClass   (const std::string& packageName,
                                             const std::string& className,
                                             uint8_t*    md5Sum,
                                             ManagementObject::writeSchemaCall_t schemaCall);
    QPID_BROKER_EXTERN void registerEvent   (const std::string& packageName,
                                             const std::string& eventName,
                                             uint8_t*    md5Sum,
                                             ManagementObject::writeSchemaCall_t schemaCall);
    QPID_BROKER_EXTERN ObjectId addObject   (ManagementObject* object,
                                             uint64_t          persistId = 0,
                                             bool              persistent = false);
    QPID_BROKER_EXTERN ObjectId addObject   (ManagementObject*  object,
                                             const std::string& key,
                                             bool               persistent = false);
    QPID_BROKER_EXTERN void raiseEvent(const ManagementEvent& event,
                                       severity_t severity = SEV_DEFAULT);
    QPID_BROKER_EXTERN void clientAdded     (const std::string& routingKey);

    QPID_BROKER_EXTERN void clusterUpdate();

    bool dispatchCommand (qpid::broker::Deliverable&       msg,
                          const std::string&         routingKey,
                          const framing::FieldTable* args,
                          const bool topic,
                          int qmfVersion);

    /** Disallow a method. Attempts to call it will receive an exception with message. */
    void disallow(const std::string& className, const std::string& methodName, const std::string& message);

    /** Disallow all QMFv1 methods (used in clustered brokers). */
    void disallowV1Methods() { disallowAllV1Methods = true; }

    /** Serialize my schemas as a binary blob into schemaOut */
    void exportSchemas(std::string& schemaOut);

    /** Serialize my remote-agent map as a binary blob into agentsOut */
    void exportAgents(std::string& agentsOut);

    /** Decode a serialized schemas and add to my schema cache */
    void importSchemas(framing::Buffer& inBuf);

    /** Decode a serialized agent map */
    void importAgents(framing::Buffer& inBuf);

    // these are in support of the managementSetup-state stuff, for synch'ing clustered brokers
    uint64_t getNextObjectId(void) { return nextObjectId; }
    void setNextObjectId(uint64_t o) { nextObjectId = o; }

    uint16_t getBootSequence(void) { return bootSequence; }
    void setBootSequence(uint16_t b) { bootSequence = b; writeData(); }

    const framing::Uuid& getUuid() const { return uuid; }
    void setUuid(const framing::Uuid& id) { uuid = id; writeData(); }

    // TODO: remove these when Variant API moved into common library.
    static types::Variant::Map toMap(const framing::FieldTable& from);
    static framing::FieldTable fromMap(const types::Variant::Map& from);
    static types::Variant::List toList(const framing::List& from);
    static framing::List fromList(const types::Variant::List& from);
    static boost::shared_ptr<framing::FieldValue> toFieldValue(const types::Variant& in);
    static types::Variant toVariant(const boost::shared_ptr<framing::FieldValue>& val);

    // For Clustering: management objects that have been marked as
    // "deleted", but are waiting for their last published object
    // update are not visible to the cluster replication code.  These
    // interfaces allow clustering to gather up all the management
    // objects that are deleted in order to allow all clustered
    // brokers to publish the same set of deleted objects.

    class DeletedObject {
      public:
        typedef boost::shared_ptr<DeletedObject> shared_ptr;
        DeletedObject(ManagementObject *, bool v1, bool v2);
        DeletedObject( const std::string &encoded );
        ~DeletedObject() {};
        void encode( std::string& toBuffer );
        const std::string getKey() const {
            // used to batch up objects of the same class type
            return std::string(packageName + std::string(":") + className);
        }

      private:
        friend class ManagementAgent;

        std::string packageName;
        std::string className;
        std::string objectId;

        std::string encodedV1Config;    // qmfv1 properties
        std::string encodedV1Inst;      // qmfv1 statistics
        qpid::types::Variant::Map encodedV2;
    };

    typedef std::vector<DeletedObject::shared_ptr> DeletedObjectList;

    /** returns a snapshot of all currently deleted management objects. */
    void exportDeletedObjects( DeletedObjectList& outList );

    /** Import a list of deleted objects to send on next publish interval. */
    void importDeletedObjects( const DeletedObjectList& inList );

private:
    struct Periodic : public qpid::sys::TimerTask
    {
        ManagementAgent& agent;

        Periodic (ManagementAgent& agent, uint32_t seconds);
        virtual ~Periodic ();
        void fire ();
    };

    //  Storage for tracking remote management agents, attached via the client
    //  management agent API.
    //
    struct RemoteAgent : public Manageable
    {
        ManagementAgent&  agent;
        uint32_t          brokerBank;
        uint32_t          agentBank;
        std::string       routingKey;
        ObjectId          connectionRef;
        qmf::org::apache::qpid::broker::Agent*    mgmtObject;
        RemoteAgent(ManagementAgent& _agent) : agent(_agent), mgmtObject(0) {}
        ManagementObject* GetManagementObject (void) const { return mgmtObject; }

        virtual ~RemoteAgent ();
        void mapEncode(qpid::types::Variant::Map& _map) const;
        void mapDecode(const qpid::types::Variant::Map& _map);
    };

    typedef std::map<ObjectId, boost::shared_ptr<RemoteAgent> > RemoteAgentMap;

    //  Storage for known schema classes:
    //
    //  SchemaClassKey     -- Key elements for map lookups
    //  SchemaClassKeyComp -- Comparison class for SchemaClassKey
    //  SchemaClass        -- Non-key elements for classes
    //
    struct SchemaClassKey
    {
        std::string name;
        uint8_t     hash[16];

        void mapEncode(qpid::types::Variant::Map& _map) const;
        void mapDecode(const qpid::types::Variant::Map& _map);
        void encode(framing::Buffer& buffer) const;
        void decode(framing::Buffer& buffer);
        uint32_t encodedBufSize() const;
    };

    struct SchemaClassKeyComp
    {
        bool operator() (const SchemaClassKey& lhs, const SchemaClassKey& rhs) const
        {
            if (lhs.name != rhs.name)
                return lhs.name < rhs.name;
            else
                for (int i = 0; i < 16; i++)
                    if (lhs.hash[i] != rhs.hash[i])
                        return lhs.hash[i] < rhs.hash[i];
            return false;
        }
    };


    struct SchemaClass
    {
        uint8_t  kind;
        ManagementObject::writeSchemaCall_t writeSchemaCall;
        std::string data;
        uint32_t pendingSequence;

        SchemaClass(uint8_t _kind=0, uint32_t seq=0) :
            kind(_kind), writeSchemaCall(0), pendingSequence(seq) {}
        SchemaClass(uint8_t _kind, ManagementObject::writeSchemaCall_t call) :
            kind(_kind), writeSchemaCall(call), pendingSequence(0) {}
        bool hasSchema () { return (writeSchemaCall != 0) || !data.empty(); }
        void appendSchema (framing::Buffer& buf);

        void mapEncode(qpid::types::Variant::Map& _map) const;
        void mapDecode(const qpid::types::Variant::Map& _map);
    };

    typedef std::map<SchemaClassKey, SchemaClass, SchemaClassKeyComp> ClassMap;
    typedef std::map<std::string, ClassMap> PackageMap;

    RemoteAgentMap               remoteAgents;
    PackageMap                   packages;

    //
    // Protected by userLock
    //
    ManagementObjectMap          managementObjects;

    //
    // Protected by addLock
    //
    ManagementObjectVector       newManagementObjects;

    framing::Uuid                uuid;

    //
    // Lock hierarchy:  If a thread needs to take both addLock and userLock,
    // it MUST take userLock first, then addLock.
    //
    sys::Mutex userLock;
    sys::Mutex addLock;

    qpid::broker::Exchange::shared_ptr mExchange;
    qpid::broker::Exchange::shared_ptr dExchange;
    qpid::broker::Exchange::shared_ptr v2Topic;
    qpid::broker::Exchange::shared_ptr v2Direct;
    std::string                  dataDir;
    uint16_t                     interval;
    qpid::broker::Broker*        broker;
    qpid::sys::Timer*            timer;
    uint16_t                     bootSequence;
    uint32_t                     nextObjectId;
    uint32_t                     brokerBank;
    uint32_t                     nextRemoteBank;
    uint32_t                     nextRequestSequence;
    bool                         clientWasAdded;
    const qpid::sys::AbsTime     startTime;
    bool                         suppressed;

    typedef std::pair<std::string,std::string> MethodName;
    typedef std::map<MethodName, std::string> DisallowedMethods;
    DisallowedMethods disallowed;
    bool disallowAllV1Methods;

    // Agent name and address
    qpid::types::Variant::Map attrMap;
    std::string       name_address;
    std::string vendorNameKey;  // "." --> "_"
    std::string productNameKey; // "." --> "_"
    std::string instanceNameKey; // "." --> "_"

    // supported management protocol
    bool qmf1Support;
    bool qmf2Support;

    // Maximum # of objects allowed in a single V2 response
    // message.
    uint32_t maxReplyObjs;

    // list of objects that have been deleted, but have yet to be published
    // one final time.
    // Indexed by a string composed of the object's package and class name.
    // Protected by userLock.
    typedef std::map<std::string, DeletedObjectList> PendingDeletedObjsMap;
    PendingDeletedObjsMap pendingDeletedObjs;

#   define MA_BUFFER_SIZE 65536
    char inputBuffer[MA_BUFFER_SIZE];
    char outputBuffer[MA_BUFFER_SIZE];
    char eventBuffer[MA_BUFFER_SIZE];
    framing::ResizableBuffer msgBuffer;

    void writeData ();
    void periodicProcessing (void);
    void deleteObjectNowLH(const ObjectId& oid);
    void encodeHeader       (framing::Buffer& buf, uint8_t  opcode, uint32_t  seq = 0);
    bool checkHeader        (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
    void sendBufferLH(framing::Buffer&             buf,
                      uint32_t                     length,
                      qpid::broker::Exchange::shared_ptr exchange,
                      const std::string&           routingKey);
    void sendBufferLH(framing::Buffer&             buf,
                      uint32_t                     length,
                      const std::string&           exchange,
                      const std::string&           routingKey);
    void sendBufferLH(const std::string&     data,
                      const std::string&     cid,
                      const qpid::types::Variant::Map& headers,
                      const std::string&     content_type,
                      qpid::broker::Exchange::shared_ptr exchange,
                      const std::string& routingKey,
                      uint64_t ttl_msec = 0);
    void sendBufferLH(const std::string& data,
                      const std::string& cid,
                      const qpid::types::Variant::Map& headers,
                      const std::string& content_type,
                      const std::string& exchange,
                      const std::string& routingKey,
                      uint64_t ttl_msec = 0);
    void moveNewObjectsLH();
    bool moveDeletedObjectsLH();

    bool authorizeAgentMessageLH(qpid::broker::Message& msg);
    void dispatchAgentCommandLH(qpid::broker::Message& msg, bool viaLocal=false);

    PackageMap::iterator findOrAddPackageLH(std::string name);
    void addClassLH(uint8_t                      kind,
                    PackageMap::iterator         pIter,
                    const std::string&           className,
                    uint8_t*                     md5Sum,
                    ManagementObject::writeSchemaCall_t schemaCall);
    void encodePackageIndication (framing::Buffer&     buf,
                                  PackageMap::iterator pIter);
    void encodeClassIndication (framing::Buffer&     buf,
                                const std::string packageName,
                                const struct SchemaClassKey key,
                                uint8_t kind);
    bool     bankInUse (uint32_t bank);
    uint32_t allocateNewBank ();
    uint32_t assignBankLH (uint32_t requestedPrefix);
    void deleteOrphanedAgentsLH();
    void sendCommandCompleteLH(const std::string& replyToKey, uint32_t sequence,
                              uint32_t code = 0, const std::string& text = "OK");
    void sendExceptionLH(const std::string& rte, const std::string& rtk, const std::string& cid, const std::string& text, uint32_t code=1, bool viaLocal=false);
    void handleBrokerRequestLH  (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handlePackageQueryLH   (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handlePackageIndLH     (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handleClassQueryLH     (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handleClassIndLH       (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handleSchemaRequestLH  (framing::Buffer& inBuffer, const std::string& replyToEx, const std::string& replyToKey, uint32_t sequence);
    void handleSchemaResponseLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handleAttachRequestLH  (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
    void handleGetQueryLH       (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
    void handleMethodRequestLH  (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
    void handleGetQueryLH       (const std::string& body, const std::string& replyToEx, const std::string& replyToKey, const std::string& cid, bool viaLocal);
    void handleMethodRequestLH  (const std::string& body, const std::string& replyToEx, const std::string& replyToKey, const std::string& cid, const qpid::broker::ConnectionToken* connToken, bool viaLocal);
    void handleLocateRequestLH  (const std::string& body, const std::string& replyToEx, const std::string &replyToKey, const std::string& cid);


    size_t validateSchema(framing::Buffer&, uint8_t kind);
    size_t validateTableSchema(framing::Buffer&);
    size_t validateEventSchema(framing::Buffer&);
    ManagementObjectMap::iterator numericFind(const ObjectId& oid);

    std::string summarizeAgents();
    void debugSnapshot(const char* title);
};

void setManagementExecutionContext(const qpid::broker::ConnectionState*);
const qpid::broker::ConnectionState* getManagementExecutionContext();
}}

#endif  /*!_ManagementAgent_*/