Merge pull request ceph#60942 from AliMasarweh/wip-alimasa-bucket-logging-filter

RGW/logging: add filtering for bucket logging 

Reviewed-by: yuvalif<[email protected]>

Author: AliMasarweh

doc/radosgw/bucket_logging.rst

@@ -40,8 +40,8 @@ Journal
 If logging type is set to "Journal", the records are written to the log bucket before the bucket operation is completed. 
 This means that if the logging action fails, the operation will not be executed, and an error will be returned to the client.
 An exception to the above are "multi/delete" log records: if writing these log records fail, the operation continues and may still be successful.
-Note that it may happen that the log records were successfully written, but the bucket operation failed, since the logs are written
-before such a failure, there will be no indication for that in the log records. 
+Journal mode supports filtering out records based on matches of the prefixes and suffixes of the logged object keys. Regular-expression matching can also be used on these to create filters.
+Note that it may happen that the log records were successfully written, but the bucket operation failed, since the logs are written.
 
 
 Bucket Logging REST API

doc/radosgw/s3/bucketops.rst

@@ -751,6 +751,14 @@ Parameters are XML encoded in the body of the request, in the following format:
       <TargetPrefix>string</TargetPrefix>
       <LoggingType>Standard|Journal</LoggingType>
       <ObjectRollTime>integer</ObjectRollTime>
+      <Filter>
+        <S3Key>
+          <FilterRule>
+            <Name>suffix/prefix/regex</Name>
+            <Value></Value>
+          </FilterRule>
+        </S3Key>
+      </Filter>
     </LoggingEnabled>
   </BucketLoggingStatus>
 
@@ -881,6 +889,14 @@ Response is XML encoded in the body of the request, in the following format:
       <TargetPrefix>string</TargetPrefix>
       <LoggingType>Standard|Journal</LoggingType>
       <ObjectRollTime>integer</ObjectRollTime>
+      <Filter>
+        <S3Key>
+          <FilterRule>
+            <Name>suffix/prefix/regex</Name>
+            <Value></Value>
+          </FilterRule>
+        </S3Key>
+      </Filter>
     </LoggingEnabled>
   </BucketLoggingStatus>
 

examples/rgw/boto3/bucket_logging.py

@@ -39,6 +39,17 @@
     },
     'ObjectRollTime': 60,
     'LoggingType': 'Journal',
+    "Filter": {
+      "Key": {
+        "FilterRules":
+        [
+          {
+            "Name": "prefix",
+            "Value": "myfile"
+          }
+        ]
+      }
+    }
   }
 }
 

examples/rgw/boto3/service-2.sdk-extras.json

@@ -310,6 +310,10 @@
                 "RecordsBatchSize":{
                     "shape":"RecordsBatchSize",
                     "documentation":"indicates how many records to batch in memory before writing to the object. if set to zero, records are written syncronously to the object. if <code>ObjectRollTime</code>e is reached, the batch of records will be written to the object regardless of the number of records. </p>"
+                },
+                "Filter":{
+                    "shape":"LoggingConfigurationFilter",
+                    "documentation":"<p>A filter for all log object. Filter for the object by its key (prefix, suffix and regex).</p>"
                 }
             },
             "documentation":"<p>Describes where logs are stored the prefix assigned to all log object keys for a bucket, and their format. also, the level the delivery guarantee of the records.</p>"
@@ -363,6 +367,18 @@
                 "Standard",
                 "Journal"
             ]
+        },
+        "LoggingConfigurationFilter":{
+            "type":"structure",
+            "members":{
+                "Key":{
+                    "shape":"S3KeyFilter",
+                    "documentation":"<p/>",
+                    "locationName":"S3Key"
+                }
+            },
+            "documentation":"<p>A filter for all log object. Filter for the object by its key (prefix, suffix and regex).</p>",
+            "locationName":"Filter"
         }
     },
     "documentation":"<p/>"

src/rgw/CMakeLists.txt

@@ -90,6 +90,7 @@ set(librgw_common_srcs
   rgw_notify_event_type.cc
   rgw_period_history.cc
   rgw_period_puller.cc
+  rgw_s3_filter.cc
   rgw_pubsub.cc
   rgw_coroutine.cc
   rgw_cr_rest.cc

src/rgw/rgw_bucket_logging.cc

@@ -31,6 +31,9 @@ bool configuration::decode_xml(XMLObj* obj) {
       logging_type = LoggingType::Standard;
     } else if (type == "Journal") {
       logging_type = LoggingType::Journal;
+      if (iter = o->find("Filter"); XMLObj* const filter_o = iter.get_next()) {
+        RGWXMLDecoder::decode_xml("S3Key", key_filter, filter_o);
+      }
     } else {
       // we don't allow for type "Any" in the configuration
       throw RGWXMLDecoder::err("invalid bucket logging record type: '" + type + "'");
@@ -73,6 +76,11 @@ void configuration::dump_xml(Formatter *f) const {
       break;
     case LoggingType::Journal:
       ::encode_xml("LoggingType", "Journal", f);
+      if (key_filter.has_content()) {
+        f->open_object_section("Filter");
+        ::encode_xml("S3Key", key_filter, f);
+        f->close_section(); // Filter
+      }
       break;
     case LoggingType::Any:
       ::encode_xml("LoggingType", "", f);
@@ -118,6 +126,10 @@ void configuration::dump(Formatter *f) const {
         break;
       case LoggingType::Journal:
         encode_json("loggingType", "Journal", f);
+        if (key_filter.has_content()) {
+          Formatter::ObjectSection s(*f, "Filter");
+          encode_json("S3Key", key_filter, f);
+        }
         break;
       case LoggingType::Any:
         encode_json("loggingType", "", f);
@@ -526,6 +538,11 @@ int log_record(rgw::sal::Driver* driver,
     if (type != LoggingType::Any && configuration.logging_type != type) {
       return 0;
     }
+    if (configuration.key_filter.has_content()) {
+      if (!match(configuration.key_filter, obj->get_name())) {
+        return 0;
+      }
+    }
     ldpp_dout(dpp, 20) << "INFO: found matching logging configuration of bucket '" << s->bucket->get_name() << 
       "' configuration: " << configuration.to_json_str() << dendl;
     if (auto ret = log_record(driver, obj, s, op_name, etag, size, configuration, dpp, y, async_completion, log_source_bucket); ret < 0) { 

src/rgw/rgw_bucket_logging.h

@@ -10,6 +10,7 @@
 #include "include/buffer.h"
 #include "include/encoding.h"
 #include "common/async/yield_context.h"
+#include "rgw_s3_filter.h"
 
 class XMLObj;
 namespace ceph { class Formatter; }
@@ -48,6 +49,14 @@ namespace rgw::bucketlogging {
       <LoggingType>Standard|Journal</LoggingType>       <!-- Ceph extension -->
       <ObjectRollTime>integer</ObjectRollTime>          <!-- Ceph extension -->
       <RecordsBatchSize>integer</RecordsBatchSize>      <!-- Ceph extension -->
+      <Filter>
+        <S3Key>
+          <FilterRule>
+            <Name>suffix/prefix/regex</Name>
+            <Value></Value>
+          </FilterRule>
+        </S3Key>
+      </Filter>
    </LoggingEnabled>
 </BucketLoggingStatus>
 */
@@ -78,6 +87,7 @@ struct configuration {
   PartitionDateSource date_source = PartitionDateSource::DeliveryTime;
   // EventTime: use only year, month, and day. The hour, minutes and seconds are set to 00 in the key
   // DeliveryTime: the time the log object was created
+  rgw_s3_key_filter key_filter;
   bool decode_xml(XMLObj *obj);
   void dump_xml(Formatter *f) const;
   void dump(Formatter *f) const; // json
@@ -92,6 +102,9 @@ struct configuration {
     encode(static_cast<int>(logging_type), bl);
     encode(records_batch_size, bl);
     encode(static_cast<int>(date_source), bl);
+    if (logging_type == LoggingType::Journal) {
+      encode(key_filter, bl);
+    }
     ENCODE_FINISH(bl);
   }
 
@@ -108,6 +121,9 @@ struct configuration {
     decode(records_batch_size, bl);
     decode(type, bl);
     date_source = static_cast<PartitionDateSource>(type);
+    if (logging_type == LoggingType::Journal) {
+      decode(key_filter, bl);
+    }
     DECODE_FINISH(bl);
   }
 };