1

"""

     2

    Full-text searching of logs

     3

"""

     4

     5

import datetime, calendar, pytz

     6

     7

import HyperEstraier as hype

     8

     9

import log_line

    10

    11

class LogSearchIndex (object) :

    12

    """

    13

        An index on the logs for a group of channels.

    14

    15

        This uses Hyper Estraier to handle searching, whereby each log line is a document (yes, I have a powerful server).

    16

    17

        These log documents have the following attributes:

    18

            @uri        - channel/date/line

    19

            @channel    - channel id

    20

            @type       - the LogType id

    21

            @timestamp  - UTC timestamp

    22

            @source     - nickname

    23

    24

        Each document then has a single line of data, which is the log message itself

    25

    """

    26

    27

    def __init__ (self, path, mode='r') :

    28

        """

    29

            Open the database, with the given mode:

    30

                r       - read-only

    31

                w       - read-write, create if not exists

    32

                a       - read-write, do not create

    33

                *       - read-write, truncate and create new

    34

        """

    35

    36

        # mapping of { mode -> flags }

    37

        mode_to_flag = {

    38

            'r':    hype.Database.DBREADER,

    39

            'w':    hype.Database.DBWRITER | hype.Database.DBCREAT,

    40

            'a':    hype.Database.DBWRITER,

    41

            '*':    hype.Database.DBWRITER | hype.Database.DBCREAT | hype.Database.DBTRUNC,

    42

        }

    43

    44

        # look up flags

    45

        flags = mode_to_flag[mode]

    46

    47

        # make instance

    48

        self.db = hype.Database()

    49

    50

        # open

    51

        if not self.db.open(path, flags) :

    52

            raise Exception("Index open failed: %s, mode=%s, flags=%#06x: %s" % (path, mode, flags, self.db.err_msg(self.db.error())))

    53

    54

    def insert (self, channel, lines) :

    55

        """

    56

            Adds a sequence of LogLines from the given LogChannel to the index, and return the number of added items

    57

        """

    58

    59

        # validate the LogChannel

    60

        assert channel.name

    61

    62

        count = 0

    63

    64

        # iterate

    65

        for line in lines :

    66

            # validate the LogLine

    67

            assert line.offset

    68

            assert line.timestamp

    69

    70

            # create new document

    71

            doc = hype.Document()

    72

    73

            # line date

    74

            date = line.timestamp.date()

    75

    76

            # convert to UTC timestamp

    77

            utc_timestamp = calendar.timegm(line.timestamp.utctimetuple())

    78

    79

            # ensure that it's not 1900

    80

            assert date.year != 1900

    81

    82

            # add URI

    83

            doc.add_attr('@uri',        "%s/%s/%d" % (channel.id, date.strftime('%Y-%m-%d'), line.offset))

    84

    85

            # add channel id

    86

            doc.add_attr('@channel',    channel.id)

    87

    88

            # add type

    89

            doc.add_attr('@type',       str(line.type))

    90

    91

            # add UTC timestamp

    92

            doc.add_attr('@timestamp',  str(utc_timestamp))

    93

    94

            # add source attribute?

    95

            if line.source :

    96

                doc.add_attr('@source', str(line.source))

    97

    98

            # add data text

    99

            doc.add_text(line.data.encode('utf8'))

   100

   101

            # put

   102

            # XXX: what does this flag mean?

   103

            if not self.db.put_doc(doc, hype.Database.PDCLEAN) :

   104

                raise Exeception("Index put_doc failed")

   105

   106

            # count

   107

            count += 1

   108

   109

        # return

   110

        return count

   111

   112

    def search_cond (self, cond) :

   113

        """

   114

            Search using a raw hype.Condition

   115

        """

   116

   117

        # execute search, unused 'flags' arg stays zero

   118

        results = self.db.search(cond, 0)

   119

   120

        # iterate over the document IDs

   121

        for doc_id in results :

   122

            # load document, this throws an exception...

   123

            # option constants are hype.Database.GDNOATTR/GDNOTEXT

   124

            doc = self.db.get_doc(doc_id, 0)

   125

   126

            # load the attributes/text

   127

            channel_id  = doc.attr('@channel')

   128

            type        = int(doc.attr('@type'))

   129

            timestamp   = datetime.datetime.fromtimestamp(int(doc.attr('@timestamp')), pytz.utc)

   130

            source      = doc.attr('@source')

   131

            data        = doc.cat_texts().decode('utf8')

   132

   133

            # build+yield to as LogLine

   134

            # XXX: ignore channel_id for now

   135

            yield log_line.LogLine(None, type, timestamp, source, data)

   136

   137

    def search (self, options=None, channel=None, phrase=None, order=None, max=None, skip=None) :

   138

        """

   139

            Search with flexible parameters

   140

   141

                options     - bitmask of hype.Condition.*

   142

                channel     - LogChannel object

   143

                phrase      - the search query phrase

   144

                order       - order attribute expression

   145

                max         - number of results to return

   146

                skip        - number of results to skip

   147

        """

   148

   149

        # build condition

   150

        cond = hype.Condition()

   151

   152

        if options :

   153

            # set options

   154

            cond.set_options(options)

   155

   156

        if channel :

   157

            # add channel attribute

   158

            cond.add_attr("@channel STREQ %s" % (channel.id, ))

   159

   160

        if phrase :

   161

            # add phrase

   162

            cond.set_phrase(phrase)

   163

   164

        if order :

   165

            # set order

   166

            cond.set_order(order)

   167

   168

        if max :

   169

            # set max

   170

            cond.set_max(max)

   171

   172

        if skip :

   173

            # set skip

   174

            cond.set_skip(skip)

   175

   176

        # execute

   177

        return self.search_cond(cond)

   178

   179

    def search_simple (self, channel, query, count=None, offset=None) :

   180

        """

   181

            Search for lines from the given channel for the given simple query

   182

        """

   183

   184

        # use search(), backwards

   185

        results = list(self.search(

   186

            # simplified phrase

   187

            options     = hype.Condition.SIMPLE,

   188

   189

            # specific channel

   190

            channel     = channel,

   191

   192

            # given phrase

   193

            phrase      = query,

   194

   195

            # order by timestamp

   196

            order       = "@timestamp NUMD",

   197

   198

            # count/offset

   199

            max         = count,

   200

            skip        = offset,

   201

        ))

   202

   203

        # reverse

   204

        return reversed(results)

   205