Updated the Tailpipe SDK to version v0.8.0 and removed the TpIndex value assignments from the tables

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/stoewer/go-strcase v1.3.0
 	github.com/turbot/go-kit v1.3.0
 	github.com/turbot/pipe-fittings/v2 v2.5.0
-	github.com/turbot/tailpipe-plugin-sdk v0.7.2
+	github.com/turbot/tailpipe-plugin-sdk v0.8.0
 	golang.org/x/sync v0.15.0
 )
 
@@ -125,7 +125,7 @@ require (
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jtolds/gls v4.20.0+incompatible // indirect
 	github.com/karrick/gows v0.3.0 // indirect
-	github.com/klauspost/compress v1.17.11 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect

go.sum

@@ -572,8 +572,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/asmfmt v1.3.2 h1:4Ri7ox3EwapiOjCki+hw14RyKk201CN4rzyCJRFLpK4=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=
 github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
-github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
-github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=
 github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -729,8 +729,8 @@ github.com/turbot/pipe-fittings/v2 v2.5.0 h1:kbATVh3GP7+p1mCboZBXw/sFEdogfNmtjPy
 github.com/turbot/pipe-fittings/v2 v2.5.0/go.mod h1:szte433cBDCaZcGe5zMVGG7uTl9HMaEYaQmuvzZRYIQ=
 github.com/turbot/pipes-sdk-go v0.12.0 h1:esbbR7bALa5L8n/hqroMPaQSSo3gNM/4X0iTmHa3D6U=
 github.com/turbot/pipes-sdk-go v0.12.0/go.mod h1:Mb+KhvqqEdRbz/6TSZc2QWDrMa5BN3E4Xw+gPt2TRkc=
-github.com/turbot/tailpipe-plugin-sdk v0.7.2 h1:2V2qccwSLr9hPT7xNs9WQu1KuY2BXPQgZczlry8A3QE=
-github.com/turbot/tailpipe-plugin-sdk v0.7.2/go.mod h1:is7Hh1u0yZS2nJ2HXwV9BpDMpuDycIt/gJtE10ufMss=
+github.com/turbot/tailpipe-plugin-sdk v0.8.0 h1:WsH0OZREJbP7+ZXh7Ty9omSn5tqoEM85LnWUJ45wci0=
+github.com/turbot/tailpipe-plugin-sdk v0.8.0/go.mod h1:kpvafTVw6KUx/kpFMshbzQLuZ6ApdWMS5ZqYQzp1q/A=
 github.com/turbot/terraform-components v0.0.0-20231213122222-1f3526cab7a7 h1:qDMxFVd8Zo0rIhnEBdCIbR+T6WgjwkxpFZMN8zZmmjg=
 github.com/turbot/terraform-components v0.0.0-20231213122222-1f3526cab7a7/go.mod h1:5hzpfalEjfcJWp9yq75/EZoEu2Mzm34eJAPm3HOW2tw=
 github.com/ulikunitz/xz v0.5.10 h1:t92gobL9l3HE202wg3rlk19F6X+JOxl9BBrCCMYEYd8=

sources/cloudwatch_log_group/cloudwatch_log_group_source.go

@@ -262,6 +262,7 @@ func (s *AwsCloudWatchLogGroupSource) filterLogEvents(ctx context.Context, input
 	paginator := cloudwatchlogs.NewFilterLogEventsPaginator(s.client, input)
 	for paginator.HasMorePages() {
 		output, err := paginator.NextPage(ctx)
+		slog.Debug("Filtering log events ===>>>>", "log_group", s.Config.LogGroupName, "output", len(output.Events))
 		if err != nil {
 			return nil, err
 		}

tables/alb_access_log/alb_access_log_table.go

@@ -1,7 +1,6 @@
 package alb_access_log
 
 import (
-	"strings"
 	"time"
 
 	"github.com/rs/xid"
@@ -61,7 +60,6 @@ func (c *AlbAccessLogTable) EnrichRow(row *AlbAccessLog, sourceEnrichmentFields
 	row.TpIngestTimestamp = time.Now()
 	row.TpTimestamp = row.Timestamp
 	row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
-	row.TpIndex = strings.TrimPrefix(row.Elb, "app/")
 
 	row.TpSourceIP = &row.ClientIP
 	row.TpIps = append(row.TpIps, row.ClientIP)

tables/alb_connection_log/alb_connection_log_table.go

@@ -61,9 +61,6 @@ func (c *AlbConnectionLogTable) EnrichRow(row *AlbConnectionLog, sourceEnrichmen
 	row.TpTimestamp = row.Timestamp
 	row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
 
-	// tp_index
-	row.TpIndex = schema.DefaultIndex
-
 	row.TpSourceIP = &row.ClientIP
 	row.TpIps = append(row.TpIps, row.ClientIP)
 	return row, nil

tables/clb_access_log/clb_access_log.go

@@ -145,7 +145,6 @@ func (c *ClbAccessLog) GetColumnDescriptions() map[string]string {
 		"user_agent":               "A User-Agent string that identifies the client that originated the request.",
 
 		// Tailpipe-specific metadata fields
-		"tp_index":          "The name of the load balancer.",
 		"tp_source_ip":      "The IP address of the requesting client.",
 		"tp_ips":            "The IP addresses of the requesting client and the registered instance that processed the request.",
 		"tp_destination_ip": "The IP address of the registered instance that processed the request.",

tables/clb_access_log/clb_access_log_table.go

@@ -58,8 +58,6 @@ func (c *ClbAccessLogTable) EnrichRow(row *ClbAccessLog, sourceEnrichmentFields
 	row.TpTimestamp = row.Timestamp
 	row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
 
-	row.TpIndex = row.Elb
-
 	row.TpSourceIP = &row.ClientIP
 	row.TpIps = append(row.TpIps, row.ClientIP)
 	if row.BackendIP != nil {

tables/cloudtrail_log/cloudtrail_log.go

@@ -128,7 +128,6 @@ func (c *CloudTrailLog) GetColumnDescriptions() map[string]string {
 
 		// Override table specific tp_* column descriptions
 		"tp_akas":      "Resource ARNs associated with the event.",
-		"tp_index":     "The AWS account ID that received the request.",
 		"tp_ips":       "IP addresses associated with the event, including the source IP address.",
 		"tp_timestamp": "The date and time the event occurred, in ISO 8601 format.",
 		"tp_usernames": "Usernames or access key IDs associated with the event.",

tables/cloudtrail_log/cloudtrail_log_table.go

@@ -87,8 +87,6 @@ func (t *CloudTrailLogTable) EnrichRow(row *CloudTrailLog, sourceEnrichmentField
 		row.TpUsernames = append(row.TpUsernames, *row.UserIdentity.UserName)
 	}
 
-	// Hive fields
-	row.TpIndex = row.RecipientAccountId
 	// convert to date in format yy-mm-dd
 	row.TpDate = row.EventTime.Truncate(24 * time.Hour)
 

tables/cost_and_usage_focus/cost_and_usage_focus_table.go

@@ -59,8 +59,6 @@ func (c *CostUsageFocusTable) EnrichRow(row *CostUsageFocus, sourceEnrichmentFie
 	row.TpID = xid.New().String()
 	row.TpIngestTimestamp = time.Now()
 
-	row.TpIndex = schema.DefaultIndex
-
 	if row.ChargePeriodStart != nil {
 		row.TpTimestamp = *row.ChargePeriodStart
 		row.TpDate = row.ChargePeriodStart.Truncate(24 * time.Hour)

tables/cost_and_usage_report/cost_and_usage_report_table.go

@@ -114,8 +114,6 @@ func (t *CostUsageReportTable) EnrichRow(row *CostUsageReport, sourceEnrichmentF
 		row.TpDate = row.BillBillingPeriodEndDate.Truncate(24 * time.Hour)
 	}
 
-	row.TpIndex = schema.DefaultIndex
-
 	return row, nil
 }
 

tables/cost_optimization_recommendation/cost_optimization_recommendation_table.go

@@ -66,9 +66,6 @@ func (t *CostOptimizationRecommendationsTable) EnrichRow(row *CostOptimizationRe
 	// convert to date in format yyyy-mm-dd
 	row.TpDate = row.LastRefreshTimestamp.Truncate(24 * time.Hour)
 
-	// TpIndex
-	row.TpIndex = schema.DefaultIndex
-
 	if row.ResourceARN != nil {
 		row.TpAkas = append(row.TpAkas, *row.ResourceARN)
 	}

tables/guardduty_finding/guardduty_finding.go

@@ -135,7 +135,6 @@ func (c *GuardDutyFinding) GetColumnDescriptions() map[string]string {
 
 		// Tailpipe-specific metadata fields
 		"tp_akas":      "The Amazon Resource Names (ARNs) associated with the finding.",
-		"tp_index":     "The AWS account ID where the finding was generated.",
 		"tp_usernames": "Usernames associated with the finding, including IAM users and access key IDs.",
 	}
 }

tables/guardduty_finding/guardduty_finding_table.go

@@ -55,7 +55,6 @@ func (c *GuardDutyFindingTable) EnrichRow(row *GuardDutyFinding, sourceEnrichmen
 	row.TpTimestamp = row.CreatedAt
 	row.TpIngestTimestamp = time.Now()
 	row.TpDate = row.CreatedAt.Truncate(24 * time.Hour)
-	row.TpIndex = *row.AccountId
 
 	row.TpAkas = append(row.TpAkas, *row.Arn)
 

tables/nlb_access_log/nlb_access_log.go

@@ -155,7 +155,6 @@ func (l *NlbAccessLog) GetColumnDescriptions() map[string]string {
 
 		// Tailpipe-specific metadata fields
 		"tp_akas":      "The ARN of the certificate served to the client.",
-		"tp_index":     "The resource ID of the load balancer handling the request.",
 		"tp_ips":       "All IP addresses associated with the request, including the client IP and destination IP.",
 		"tp_source_ip": "The IP address of the client initiating the connection.",
 	}

tables/nlb_access_log/nlb_access_log_table.go

@@ -1,7 +1,6 @@
 package nlb_access_log
 
 import (
-	"strings"
 	"time"
 
 	"github.com/rs/xid"
@@ -58,7 +57,6 @@ func (c *NlbAccessLogTable) EnrichRow(row *NlbAccessLog, sourceEnrichmentFields
 	row.TpIngestTimestamp = time.Now()
 	row.TpTimestamp = row.Timestamp
 	row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
-	row.TpIndex = strings.TrimPrefix(row.Elb, "net/")
 
 	row.TpSourceIP = &row.ClientIP
 	row.TpIps = append(row.TpIps, row.ClientIP)

tables/s3_server_access_log/s3_server_access_log.go

@@ -158,7 +158,6 @@ func (c *S3ServerAccessLog) GetColumnDescriptions() map[string]string {
 		"version_id":          "The version ID of the object, if versioning is enabled.",
 
 		// Tailpipe-specific metadata fields
-		"tp_index":     "The name of the S3 bucket where the request was made.",
 		"tp_ips":       "All IP addresses associated with the request, including the remote IP.",
 		"tp_usernames": "Canonical user IDs or role ARNs associated with the request.",
 	}

tables/s3_server_access_log/s3_server_access_log_table.go

@@ -61,7 +61,6 @@ func (c *S3ServerAccessLogTable) EnrichRow(row *S3ServerAccessLog, sourceEnrichm
 	row.TpTimestamp = row.Timestamp
 	row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
 
-	row.TpIndex = row.Bucket
 	row.TpSourceIP = &row.RemoteIP
 	row.TpIps = append(row.TpIps, row.RemoteIP)
 

tables/vpc_flow_log/vpc_flow_log_table.go

@@ -350,9 +350,6 @@ func (c *VpcFlowLogTable) EnrichRow(row *types.DynamicRow, sourceEnrichmentField
 		row.OutputColumns[constants.TpIps] = ips
 	}
 
-	// tp_index
-	row.OutputColumns[constants.TpIndex] = schema.DefaultIndex
-
 	// tp_akas
 	var akas []string
 	if ecsClusterArn, ok := row.GetSourceValue("ecs_cluster_arn"); ok && ecsClusterArn != VpcFlowLogTableNilValue {

tables/waf_traffic_log/waf_traffic_log.go

@@ -130,7 +130,6 @@ func (c *WafTrafficLog) GetColumnDescriptions() map[string]string {
 
 		// Override table specific tp_* column descriptions
 		"tp_akas":      "List of ARNs (Amazon Resource Names) associated with the event, if applicable.",
-		"tp_index":     "The AWS Web ACL ID that processed or received the request.",
 		"tp_ips":       "IP addresses related to the request, including the source (client) IP and any intermediary addresses.",
 		"tp_timestamp": "The timestamp when the request was made, formatted in ISO 8601 (UTC).",
 	}

tables/waf_traffic_log/waf_traffic_log_table.go

@@ -80,8 +80,6 @@ func (c *WafTrafficLogTable) EnrichRow(row *WafTrafficLog, sourceEnrichmentField
 		}
 	}
 
-	row.TpIndex = *row.WebAclId
-
 	return row, nil
 }