Class: Loki::Batch

Inherits:
Object
  • Object
show all
Defined in:
lib/logstash/outputs/loki/batch.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(e) ⇒ Batch

Returns a new instance of Batch.



6
7
8
9
10
11
 
# File 'lib/logstash/outputs/loki/batch.rb', line 6

def initialize(e)
    @bytes = 0
    @createdAt = Time.now
    @streams = {}
    add(e)
end

Instance Attribute Details

#streams ⇒ Object (readonly)

Returns the value of attribute streams.



5
6
7
 
# File 'lib/logstash/outputs/loki/batch.rb', line 5

def streams
  @streams
end

Instance Method Details

#add(e) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/logstash/outputs/loki/batch.rb', line 17

def add(e)
    @bytes = @bytes + e.entry['line'].length

    # Append the entry to an already existing stream (if any)
    labels = e.labels.sort.to_h
    labelkey = labels.to_s
    if @streams.has_key?(labelkey)
        stream = @streams[labelkey]
        stream['entries'].append(e.entry)
        return
    else
        # Add the entry as a new stream
        @streams[labelkey] = {
            "labels" => labels,
            "entries" => [e.entry],
        }
    end
end

#age ⇒ Object 



40
41
42
 
# File 'lib/logstash/outputs/loki/batch.rb', line 40

def age()
    return Time.now - @createdAt
end

#build_stream(stream) ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/logstash/outputs/loki/batch.rb', line 52

def build_stream(stream)
    values = []
    stream['entries'].each { |entry|
        if entry.key?('metadata')
            sorted_metadata = entry['metadata'].sort.to_h
            values.append([
                entry['ts'].to_s,
                entry['line'],
                sorted_metadata
            ])
        else
            values.append([
                entry['ts'].to_s,
                entry['line']
            ])
        end
    }
    return {
        'stream'=>stream['labels'],
        'values' => values
    }
end

#size_bytes ⇒ Object 



13
14
15
 
# File 'lib/logstash/outputs/loki/batch.rb', line 13

def size_bytes
    return @bytes
end

#size_bytes_after(line) ⇒ Object 



36
37
38
 
# File 'lib/logstash/outputs/loki/batch.rb', line 36

def size_bytes_after(line)
    return @bytes + line.length
end

#to_json ⇒ Object 



44
45
46
47
48
49
50
 
# File 'lib/logstash/outputs/loki/batch.rb', line 44

def to_json
    streams = []
    @streams.each { |_ , stream|
        streams.append(build_stream(stream))
    }
    return {"streams"=>streams}.to_json
end