message_decompression.go

message_decompression.go
	/* 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: "versioninfo\|CLUSTERVERSION_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 }