refactor: reworked aws_vpc_flow_log table as a custom table allowing for formats to be used #137

graza-io · 2025-04-02T19:06:42Z

No description provided.

…for formats to be used

tables/vpc_flow_log/vpc_flow_log_extractor.go

graza-io · 2025-04-09T10:16:30Z

tables/vpc_flow_log/vpc_flow_log_extractor.go

@@ -32,26 +40,21 @@ func (c *VPCFlowLogExtractor) Extract(_ context.Context, a any) ([]any, error) {
 		return nil, fmt.Errorf("expected byte[], got %T", a)
 	}

-	mappedData, err := ConvertToMapSlice(string(jsonBytes))
+	mappedData, err := ConvertToMapSlice(c, string(jsonBytes))
 	if err != nil {
 		return nil, fmt.Errorf("error in mapping the results in the form of []map[string]string")
 	}

 	var res = make([]any, len(mappedData))
 	for i, record := range mappedData {


Is this iteration required, previously the mappedData was iterated as a mapping was done and errors may have been reported out, I think we can just return mappedData here?

As we are not using any mapper here, the mapped data should be of type (*types.DynamicRow) I have made the changes to return the []*types.DynamicRow{}. Otherwisewe will get the error like required *types.DynamicRow got map[string]string

tables/vpc_flow_log/vpc_flow_log_extractor.go

tables/vpc_flow_log/vpc_flow_log_format.go

graza-io · 2025-04-09T10:31:18Z

tables/vpc_flow_log/vpc_flow_log_mapper.go

I had removed the need for this mapper, I can see you've added it back purely for S3 source, any particular reason we can't just use the default approach of the regex mapper?

Ah, my bad — I missed removing it in the initial commit. We no longer need this mapper since the Extractor now handles everything. I haven’t tried using the Gonx or Regex mapper in this case. My main goal was to preserve the previous table functionality, so we used an Extractor, which parses the entire object content at once and returns the rows directly.

@ParthaI What's the effect on removing vs. keeping this mapper for current users of the table who pull flow logs from an S3 bucket today for default and custom logs?

@cbruno10, in general, the mapper is designed to work on a per-row basis, while the extractor is intended for handling bulk or batch row data. In our earlier implementation, we used an extractor to download the full object content and map the column values based on the header line (i.e., the first line of the log data). To keep the behavior consistent with that approach, I’ve continued using an extractor here.

Additionally, we don't require a custom mapper in this case because the default SDK-provided mapper for custom tables already handles the mapping logic effectively.

Here are a few considerations when comparing extractor vs. mapper for this use case:

With a mapper, the user must explicitly provide a format layout. However, with an extractor, we have the flexibility to infer the header (i.e., the format layout) from the first line of the log file—whether or not the user specifies a format block.

In some cases, such as when logs are exported from CloudWatch to an S3 bucket, the log files don’t include a header line. The current extractor-based design handles this scenario well: users can still specify a format block to define the layout manually. The default mapper would work the same way in this case if it were used, but the extractor approach makes the solution more adaptive and user-friendly.

graza-io · 2025-04-09T10:32:45Z

tables/vpc_flow_log/vpc_flow_log_table.go

+const VpcFlowLogTableSkippedData = "SKIPDATA"
+const VpcFlowLogTableNoData = "NODATA"


These are valid log entries, I don't think we should forcefully remove them in code but rather make a suggestion that users can filter them out using a filter?

Hmm, good point, @cbruno10 - could we consider keeping the SKIPDATA/NODATA values as-is instead of skipping them? That said, this behavior would then apply to all columns. We still need to retain the check for tp_timestamp and tp_data, since those are required columns in the Tailpipe DB.

Yes, I think I'd keep them, as users can filter out these values when collecting or in their queries like @graza-io mentioned. May be worth adding an example in the Example Configurations section showing how to avoid collecting these with filter.

tables/vpc_flow_log/vpc_flow_log_table.go

Copilot

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Copilot · 2025-04-09T12:11:43Z

tables/vpc_flow_log/vpc_flow_log_table.go

+
+	row.OutputColumns[constants.TpTable] = VpcFlowLogTableIdentifier
+
+	if startTime, ok := row.GetSourceValue("start_time"); ok && (startTime != VpcFlowLogTableSkippedData || startTime != VpcFlowLogTableNoData || startTime != VpcFlowLogTableNilValue) {


The condition uses logical OR (||) when checking against invalid values, which will always evaluate to true. Replace the OR operators with AND (&&) to correctly validate that startTime is not equal to any of the invalid constants.

Suggested change

if startTime, ok := row.GetSourceValue("start_time"); ok && (startTime != VpcFlowLogTableSkippedData || startTime != VpcFlowLogTableNoData || startTime != VpcFlowLogTableNilValue) {

if startTime, ok := row.GetSourceValue("start_time"); ok && (startTime != VpcFlowLogTableSkippedData && startTime != VpcFlowLogTableNoData && startTime != VpcFlowLogTableNilValue) {

Copilot · 2025-04-09T12:11:43Z

tables/vpc_flow_log/vpc_flow_log_mapper.go

-func (m *VPCFlowLogMapper) Identifier() string {
-	return "vpc_flow_log_mapper"
+func (c *VPCFlowLogMapper) Identifier() string {
+	return "aws_waf_traffic_log_mapper"


The mapper Identifier returns 'aws_waf_traffic_log_mapper', which appears inconsistent with the VPC Flow Log context. It should likely return an identifier that reflects VPC Flow Log mapping, for example 'vpc_flow_log_mapper'.

Suggested change

return "aws_waf_traffic_log_mapper"

return "vpc_flow_log_mapper"

cbruno10 · 2025-04-09T12:14:10Z

tables/vpc_flow_log/vpc_flow_log_table.go

+func (c *VpcFlowLogTable) GetTableDefinition() *schema.TableSchema {
+	return &schema.TableSchema{
+		Name: VpcFlowLogTableIdentifier,
+		Columns: []*schema.ColumnSchema{


@ParthaI Are there any changes from the previous version of the table, or are they the same?

They are the same, @cbruno10- we've just split the VpcFlowLog struct into a custom schema.

… dynamic table accept only string in it's mapper

…he doc with example config to exclude the NODATA and SKIPDATA for a column

…licable for log-status field

…g the file source implementation to the table map

…ilepath package instead of regex match

… stream name match

Copilot

Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.

Copilot · 2025-04-17T03:32:17Z

aws/plugin.go

+	// register formats
+	table.RegisterFormatPresets(vpc_flow_log.VPCFlowLogTableFormatPresets...)
+	table.RegisterFormat[*vpc_flow_log.VPCFlowLogTableFormat]()
+	row_source.RegisterRowSource[*cloudwatch_log_group.AwsCloudWatchLogGroupSource]()


The AwsCloudWatchLogGroupSource row source is registered twice, which may lead to duplicate registrations. Please remove one of the duplicate calls to avoid potential conflicts.

Suggested change

row_source.RegisterRowSource[*cloudwatch_log_group.AwsCloudWatchLogGroupSource]()

cbruno10

@ParthaI Please see comments, thanks!

cbruno10 · 2025-04-17T03:55:19Z