Add table Lambda flow log

Author: ParthaI

aws/plugin.go

@@ -12,6 +12,7 @@ import (
 	"github.com/turbot/tailpipe-plugin-aws/tables/cost_and_usage_report"
 	"github.com/turbot/tailpipe-plugin-aws/tables/cost_optimization_recommendation"
 	"github.com/turbot/tailpipe-plugin-aws/tables/guardduty_finding"
+	"github.com/turbot/tailpipe-plugin-aws/tables/lambda_log"
 	"github.com/turbot/tailpipe-plugin-aws/tables/nlb_access_log"
 	"github.com/turbot/tailpipe-plugin-aws/tables/s3_server_access_log"
 	"github.com/turbot/tailpipe-plugin-aws/tables/vpc_flow_log"
@@ -40,6 +41,7 @@ func init() {
 	table.RegisterTable[*s3_server_access_log.S3ServerAccessLog, *s3_server_access_log.S3ServerAccessLogTable]()
 	table.RegisterTable[*vpc_flow_log.VpcFlowLog, *vpc_flow_log.VpcFlowLogTable]()
 	table.RegisterTable[*waf_traffic_log.WafTrafficLog, *waf_traffic_log.WafTrafficLogTable]()
+	table.RegisterTable[*lambda_log.LambdaLog, *lambda_log.LambdaLogTable]()
 
 	// register sources
 	row_source.RegisterRowSource[*s3_bucket.AwsS3BucketSource]()

tables/lambda_log/lambda_log.go

@@ -0,0 +1,23 @@
+package lambda_log
+
+import (
+	"time"
+
+	"github.com/turbot/tailpipe-plugin-sdk/schema"
+)
+
+type LambdaLog struct {
+	schema.CommonFields
+
+	Timestamp *time.Time `json:"timestamp,omitempty"`
+	RequestID *string    `json:"request_id,omitempty"`
+	LogType   *string    `json:"log_type,omitempty"`
+	LogLevel  *string    `json:"log_level,omitempty"`
+	Message   *string    `json:"message,omitempty"`
+
+	// Report Specific Fields
+	Duration       *float64 `json:"duration,omitempty"`
+	BilledDuration *float64 `json:"billed_duration,omitempty"`
+	MemorySize     *int     `json:"memory_size,omitempty"`
+	MaxMemoryUsed  *int     `json:"max_memory_used,omitempty"`
+}

tables/lambda_log/lambda_log_mapper.go

@@ -0,0 +1,89 @@
+package lambda_log
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strconv"
+	"strings"
+	"time"
+
+	// "github.com/turbot/tailpipe-plugin-aws/rows"
+	// "github.com/turbot/tailpipe-plugin-sdk/table"
+	"github.com/turbot/tailpipe-plugin-sdk/mappers"
+)
+
+type LambdaLogMapper struct {
+}
+
+func (m *LambdaLogMapper) Identifier() string {
+	return "lambda_log_mapper"
+}
+
+func (m *LambdaLogMapper) Map(_ context.Context, a any, _ ...mappers.MapOption[*LambdaLog]) (*LambdaLog, error) {
+	row := &LambdaLog{}
+
+	rawRow := ""
+
+	switch v := a.(type) {
+	case []byte:
+		rawRow = string(v)
+	case string:
+		rawRow = v
+	case *string:
+		rawRow = *v
+	default:
+		return nil, fmt.Errorf("expected string, got %T", a)
+	}
+
+	slog.Error("rawRow ---->>>", rawRow)
+
+	rawRow = strings.TrimSuffix(rawRow, "\n")
+	fields := strings.Fields(rawRow)
+
+	switch fields[0] {
+	case "START", "END":
+		row.LogType = &fields[0]
+		row.RequestID = &fields[2]
+	case "REPORT":
+		row.LogType = &fields[0]
+		row.RequestID = &fields[2]
+		duration, err := strconv.ParseFloat(fields[4], 64)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing duration: %w", err)
+		}
+		row.Duration = &duration
+		billed, err := strconv.ParseFloat(fields[8], 64)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing billed duration: %w", err)
+		}
+		row.BilledDuration = &billed
+		mem, err := strconv.Atoi(fields[12])
+		if err != nil {
+			return nil, fmt.Errorf("error parsing memory size: %w", err)
+		}
+		row.MemorySize = &mem
+		maxMem, err := strconv.Atoi(fields[17])
+		if err != nil {
+			return nil, fmt.Errorf("error parsing max memory used: %w", err)
+		}
+		row.MaxMemoryUsed = &maxMem
+	default:
+		t := "LOG"
+		row.LogType = &t
+
+		ts, err := time.Parse(time.RFC3339, fields[0])
+		if err != nil {
+			return nil, fmt.Errorf("error parsing timestamp: %w", err)
+		}
+		row.Timestamp = &ts
+
+		row.RequestID = &fields[1]
+		row.LogLevel = &fields[2]
+		strip := fmt.Sprintf("%s%s", strings.Join(fields[:3], "\t"), "\t")
+		stripped := strings.TrimPrefix(rawRow, strip)
+		row.Message = &stripped
+	}
+
+	return row, nil
+}

tables/lambda_log/lambda_log_table.go

@@ -0,0 +1,74 @@
+package lambda_log
+
+import (
+	"time"
+
+	"github.com/rs/xid"
+	"github.com/turbot/pipe-fittings/v2/utils"
+
+	"github.com/turbot/tailpipe-plugin-aws/sources/cloudwatch_log_group"
+	"github.com/turbot/tailpipe-plugin-aws/sources/s3_bucket"
+	"github.com/turbot/tailpipe-plugin-sdk/artifact_source"
+	"github.com/turbot/tailpipe-plugin-sdk/artifact_source_config"
+	"github.com/turbot/tailpipe-plugin-sdk/constants"
+	"github.com/turbot/tailpipe-plugin-sdk/row_source"
+	"github.com/turbot/tailpipe-plugin-sdk/schema"
+	"github.com/turbot/tailpipe-plugin-sdk/table"
+)
+
+const LambdaLogTableIdentifier = "aws_lambda_log"
+
+type LambdaLogTable struct{}
+
+func (c *LambdaLogTable) Identifier() string {
+	return LambdaLogTableIdentifier
+}
+
+func (c *LambdaLogTable) GetSourceMetadata() ([]*table.SourceMetadata[*LambdaLog], error) {
+	defaultS3ArtifactConfig := &artifact_source_config.ArtifactSourceConfigImpl{
+		FileLayout: utils.ToStringPointer("AWSLogs/(%{DATA:org_id}/)?%{NUMBER:account_id}/lambda/%{DATA:function_name}/%{YEAR:year}/%{MONTHNUM:month}/%{MONTHDAY:day}/%{DATA}.log"),
+	}
+
+	return []*table.SourceMetadata[*LambdaLog]{
+		{
+			// S3 artifact source
+			SourceName: s3_bucket.AwsS3BucketSourceIdentifier,
+			Mapper:     &LambdaLogMapper{},
+			Options: []row_source.RowSourceOption{
+				artifact_source.WithDefaultArtifactSourceConfig(defaultS3ArtifactConfig),
+				artifact_source.WithRowPerLine(),
+			},
+		},
+		{
+			// S3 artifact source
+			SourceName: cloudwatch_log_group.AwsCloudwatchLogGroupSourceIdentifier,
+			Mapper:     &LambdaLogMapper{},
+		},
+		{
+			// any artifact source
+			SourceName: constants.ArtifactSourceIdentifier,
+			Mapper:     &LambdaLogMapper{},
+			Options: []row_source.RowSourceOption{
+				artifact_source.WithRowPerLine(),
+			},
+		},
+	}, nil
+}
+
+func (c *LambdaLogTable) EnrichRow(row *LambdaLog, sourceEnrichmentFields schema.SourceEnrichment) (*LambdaLog, error) {
+	row.CommonFields = sourceEnrichmentFields.CommonFields
+
+	// Record standardization
+	row.TpID = xid.New().String()
+	row.TpIngestTimestamp = time.Now()
+	if !row.TpTimestamp.IsZero() {
+		row.TpTimestamp = *row.Timestamp
+		row.TpDate = row.Timestamp.Truncate(24 * time.Hour)
+	}
+
+	row.TpIndex = schema.DefaultIndex
+
+	// TODO: Add enrichment fields
+
+	return row, nil
+}