/
message_decompression.go
86 lines (77 loc) · 2.73 KB
/
message_decompression.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
/*
Copyright © 2023 Red Hat, Inc.
Licensed 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.
*/
// Package consumer contains interface for any consumer that is able to
// process messages. It also contains implementation of Kafka consumer.
//
// It is expected that consumed messages are generated by ccx-data-pipeline
// based on OCP rules framework. The report generated by the framework are
// enhanced with more context information taken from different sources, like
// the organization ID, account number, unique cluster name, and the
// LastChecked timestamp (taken from the incoming Kafka record containing the
// URL to the archive).
//
// It is also expected that consumed messages contains one INFO rule hit that
// contains cluster version. That rule hit is produced by special rule used
// only in external data pipeline:
// "version_info|CLUSTER_VERSION_INFO"
package consumer
import (
"bytes"
"compress/gzip"
"io"
"github.com/rs/zerolog/log"
)
const (
// first magic byte of stream/file compressed by Gzip
gzipMagicNumber1 = 31
// second magic byte of stream/file compressed by Gzip
gzipMagicNumber2 = 139
)
// IsMessageInGzipFormat function checks if the format of the message is gzip
// if it is it will return true if not it will return false
func IsMessageInGzipFormat(messageValue []byte) bool {
if messageValue == nil {
return false
}
if len(messageValue) < 2 {
return false
}
// Checking for first 2 bytes in gzip stream witch are 31 and 139
// see also https://en.wikipedia.org/wiki/Gzip
if messageValue[0] == gzipMagicNumber1 && messageValue[1] == gzipMagicNumber2 {
return true
}
return false
}
// DecompressMessage will try to decompress the message if the message is compressed
// by using any supported method (GZIP at this moment)
func DecompressMessage(messageValue []byte) ([]byte, error) {
if IsMessageInGzipFormat(messageValue) {
reader := bytes.NewReader(messageValue)
gzipReader, err := gzip.NewReader(reader)
if err != nil {
return nil, err
}
defer func(r *gzip.Reader) {
if err := r.Close(); err != nil {
log.Error().Err(err).Msgf("failed to close gzip reader: %s", err.Error())
}
}(gzipReader)
decompresed, err := io.ReadAll(gzipReader)
if err != nil {
return nil, err
}
return decompresed, err
}
return messageValue, nil
}