[pkg/ottl] Add "NetworkDirection" converter

.chloggen/add_network_direction_converter.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/ottl
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: NetworkDirection calculates the network direction given a source IP address, destination IP address, and a list of internal networks.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [31930]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

pkg/ottl/e2e/e2e_test.go

@@ -600,6 +600,12 @@ func Test_e2e_converters(t *testing.T) {
 				tCtx.GetLogRecord().Attributes().PutInt("test", 1000000)
 			},
 		},
+		{
+			statement: `set(attributes["test"], NetworkDirection("192.168.1.1", "192.168.1.2", ["private"]))`,
+			want: func(tCtx ottllog.TransformContext) {
+				tCtx.GetLogRecord().Attributes().PutStr("test", "internal")
+			},
+		},
 		{
 			statement: `set(attributes["test"], "pass") where Now() - Now() < Duration("1h")`,
 			want: func(tCtx ottllog.TransformContext) {

pkg/ottl/ottlfuncs/README.md

@@ -445,6 +445,7 @@ Available Converters:
 - [Minutes](#minutes)
 - [Month](#month)
 - [Nanoseconds](#nanoseconds)
+- [NetworkDirection](#networkdirection)
 - [Now](#now)
 - [ParseCSV](#parsecsv)
 - [ParseJSON](#parsejson)
@@ -1239,6 +1240,32 @@ Examples:
 
 - `Nanoseconds(Duration("1h"))`
 
+### NetworkDirection
+
+`NetworkDirection(sourceIP, destinationIP, Optional[[]internal_networks])`
+
+`NetworkDirection` function calculates the network direction given a source IP address, destination IP address, and a list of internal networks.
+Returned value is one of following strings: `internal`,`external`,`inbound`,`outbound`.
+
+`sourceIP` is a getter that returns string, `destinationIP` is a getter that returns string and `internal_networks` is an optional array of strings.
+
+The named ranges supported for the `internal_networks` option are:
+
+- `loopback` - Matches loopback addresses in the range of 127.0.0.0/8 or ::1/128.
+- `unicast` or `global_unicast` - Matches global unicast addresses defined in RFC 1122, RFC 4632, and RFC 4291 with the exception of the IPv4 broadcast address (255.255.255.255). This includes private address ranges.
+- `multicast` - Matches multicast addresses.
+- `interface_local_multicast` - Matches IPv6 interface-local multicast addresses.
+- `link_local_unicast` - Matches link-local unicast addresses.
+- `link_local_multicast` - Matches link-local multicast addresses.
+- `private` - Matches private address ranges defined in RFC 1918 (IPv4) and RFC 4193 (IPv6).
+- `public` - Matches addresses that are not loopback, unspecified, IPv4 broadcast, link local unicast, link local multicast, interface local multicast, or private.
+- `unspecified` - Matches unspecified addresses (either the IPv4 address "0.0.0.0" or the IPv6 address "::").
+
+- `NetworkDirection("192.168.1.1", "192.168.1.2", ["private"])`
+- `NetworkDirection("192.168.1.1", "192.168.1.2", ["public"])`
+- `NetworkDirection("0.0.0.0", "0.0.0.0", ["unspecified"])`
+- `NetworkDirection("192.168.1.1", "10.0.1.1", ["10.0.0.0/8"])`
+
 ### Now
 
 `Now()`

pkg/ottl/ottlfuncs/func_network_direction.go

@@ -0,0 +1,186 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ottlfuncs // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs"
+
+import (
+	"context"
+	"fmt"
+	"net"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+)
+
+const (
+	// direction
+	directionInternal = "internal"
+	directionExternal = "external"
+	directionInbound  = "inbound"
+	directionOutbound = "outbound"
+
+	// netwroks
+	loopbackNamedNetwork           = "loopback"
+	globalUnicastNamedNetwork      = "global_unicast"
+	unicastNamedNetwork            = "unicast"
+	linkLocalUnicastNamedNetwork   = "link_local_unicast"
+	interfaceLocalNamedNetwork     = "interface_local_multicast"
+	linkLocalMulticastNamedNetwork = "link_local_multicast"
+	multicastNamedNetwork          = "multicast"
+	unspecifiedNamedNetwork        = "unspecified"
+	privateNamedNetwork            = "private"
+	publicNamedNetwork             = "public"
+)
+
+type NetworkDirectionArguments[K any] struct {
+	SourceIP         ottl.StringGetter[K]
+	DestinationIP    ottl.StringGetter[K]
+	InternalNetworks ottl.Optional[[]string]
+}
+
+func NewNetworkDirectionFactory[K any]() ottl.Factory[K] {
+	return ottl.NewFactory("NetworkDirection", &NetworkDirectionArguments[K]{}, createNetworkDirectionFunction[K])
+}
+
+func createNetworkDirectionFunction[K any](_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[K], error) {
+	args, ok := oArgs.(*NetworkDirectionArguments[K])
+	if !ok {
+		return nil, fmt.Errorf("URLFactory args must be of type *NetworkDirectionArguments[K]")
+	}
+
+	return networkDirection(args.SourceIP, args.DestinationIP, args.InternalNetworks), nil //revive:disable-line:var-naming
+}
+
+func networkDirection[K any](sourceIP ottl.StringGetter[K], destinationIP ottl.StringGetter[K], internalNetworksOptional ottl.Optional[[]string]) ottl.ExprFunc[K] { //revive:disable-line:var-naming
+	var internalNetworks []string
+	if !internalNetworksOptional.IsEmpty() {
+		for _, network := range internalNetworksOptional.Get() {
+			if len(network) > 0 {
+				internalNetworks = append(internalNetworks, network)
+			}
+		}
+	}
+
+	return func(ctx context.Context, tCtx K) (any, error) {
+		sourceIPString, err := sourceIP.Get(ctx, tCtx)
+		if err != nil {
+			return nil, fmt.Errorf("failed parsing source IP: %w", err)
+		}
+
+		if sourceIPString == "" {
+			return nil, fmt.Errorf("source IP cannot be empty")
+		}
+
+		destinationIPString, err := destinationIP.Get(ctx, tCtx)
+		if err != nil {
+			return nil, fmt.Errorf("failed parsing destination IP: %w", err)
+		}
+
+		if destinationIPString == "" {
+			return nil, fmt.Errorf("destination IP cannot be empty")
+		}
+
+		sourceAddress := net.ParseIP(sourceIPString)
+		if sourceAddress == nil {
+			return nil, fmt.Errorf("source IP %q is not a valid address", sourceIPString)
+		}
+
+		sourceInternal, err := isInternalIP(sourceAddress, internalNetworks)
+		if err != nil {
+			return nil, fmt.Errorf("failed determining whether source IP is internal: %w", err)
+		}
+
+		destinationAddress := net.ParseIP(destinationIPString)
+		if destinationAddress == nil {
+			return nil, fmt.Errorf("destination IP %q is not a valid address", destinationIPString)
+		}
+
+		destinationInternal, err := isInternalIP(destinationAddress, internalNetworks)
+		if err != nil {
+			return nil, fmt.Errorf("failed determining whether destination IP is internal: %w", err)
+		}
+
+		if sourceInternal && destinationInternal {
+			return directionInternal, nil
+		}
+		if sourceInternal {
+			return directionOutbound, nil
+		}
+		if destinationInternal {
+			return directionInbound, nil
+		}
+		return directionExternal, nil
+	}
+}
+
+func isInternalIP(addr net.IP, networks []string) (bool, error) {
+	for _, network := range networks {
+		inNetwork, err := isIPInNetwork(addr, network)
+		if err != nil {
+			return false, err
+		}
+
+		if inNetwork {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func isIPInNetwork(addr net.IP, network string) (bool, error) {
+	switch network {
+	case loopbackNamedNetwork:
+		return addr.IsLoopback(), nil
+	case globalUnicastNamedNetwork:
+		return addr.IsGlobalUnicast(), nil
+	case linkLocalUnicastNamedNetwork:
+		return addr.IsLinkLocalUnicast(), nil
+	case linkLocalMulticastNamedNetwork:
+		return addr.IsLinkLocalMulticast(), nil
+	case interfaceLocalNamedNetwork:
+		return addr.IsInterfaceLocalMulticast(), nil
+	case multicastNamedNetwork:
+		return addr.IsMulticast(), nil
+	case privateNamedNetwork:
+		return isPrivateNetwork(addr), nil
+	case publicNamedNetwork:
+		return isPublicNetwork(addr), nil
+	case unspecifiedNamedNetwork:
+		return addr.IsUnspecified(), nil
+	}
+	// cidr range
+	return isInRange(addr, network)
+}
+
+func isPrivateNetwork(addr net.IP) bool {
+	isAddrInRange, _ := isInRange(addr, "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8")
+	return isAddrInRange
+}
+
+func isPublicNetwork(addr net.IP) bool {
+	if isPrivateNetwork(addr) ||
+		addr.IsLoopback() ||
+		addr.IsUnspecified() ||
+		addr.IsLinkLocalUnicast() ||
+		addr.IsLinkLocalMulticast() ||
+		addr.IsInterfaceLocalMulticast() ||
+		addr.Equal(net.IPv4bcast) {
+		return false
+	}
+
+	return true
+}
+
+func isInRange(addr net.IP, networks ...string) (bool, error) {
+	for _, network := range networks {
+		_, mask, err := net.ParseCIDR(network)
+		if err != nil {
+			return false, fmt.Errorf("invalid network definition for %q", network)
+		}
+
+		if mask.Contains(addr) {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}

pkg/ottl/ottlfuncs/func_network_direction_test.go

@@ -0,0 +1,76 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ottlfuncs
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+)
+
+func Test_NetworkDirection(t *testing.T) {
+	testCases := []struct {
+		sourceIP          string
+		destinationIP     string
+		internalNetworks  []string
+		expectedDirection string
+		expectedError     string
+	}{
+		// cidr range tests
+		{"10.0.1.1", "192.168.1.2", []string{"10.0.0.0/8"}, "outbound", ""},
+		{"192.168.1.2", "10.0.1.1", []string{"10.0.0.0/8"}, "inbound", ""},
+
+		// private network tests
+		{"192.168.1.1", "192.168.1.2", []string{"private"}, "internal", ""},
+		{"10.0.1.1", "192.168.1.2", []string{"private"}, "internal", ""},
+		{"192.168.1.1", "172.16.0.1", []string{"private"}, "internal", ""},
+		{"192.168.1.1", "fd12:3456:789a:1::1", []string{"private"}, "internal", ""},
+
+		// public network tests
+		{"192.168.1.1", "192.168.1.2", []string{"public"}, "external", ""},
+		{"10.0.1.1", "192.168.1.2", []string{"public"}, "external", ""},
+		{"192.168.1.1", "172.16.0.1", []string{"public"}, "external", ""},
+		{"192.168.1.1", "fd12:3456:789a:1::1", []string{"public"}, "external", ""},
+
+		// unspecified tests
+		{"0.0.0.0", "0.0.0.0", []string{"unspecified"}, "internal", ""},
+		{"::", "::", []string{"unspecified"}, "internal", ""},
+
+		// invalid inputs tests
+		{"invalid", "192.168.1.2", []string{}, "", `source IP "invalid" is not a valid address`},
+		{"192.168.1.1", "invalid", []string{}, "", `destination IP "invalid" is not a valid address`},
+		{"192.168.1.1", "192.168.1.2", []string{"10.0.0.0/8", "invalid"}, "", `failed determining whether source IP is internal: invalid network definition for "invalid"`},
+	}
+
+	for i, testCase := range testCases {
+		t.Run(fmt.Sprintf("test case #%d: %s -> %s", i, testCase.sourceIP, testCase.destinationIP), func(t *testing.T) {
+			internalNetworksOptional := ottl.NewTestingOptional[[]string](testCase.internalNetworks)
+
+			source := &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return testCase.sourceIP, nil
+				},
+			}
+
+			destination := &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return testCase.destinationIP, nil
+				},
+			}
+			exprFunc := networkDirection(source, destination, internalNetworksOptional)
+			result, err := exprFunc(context.Background(), nil)
+			if len(testCase.expectedError) > 0 {
+				assert.Error(t, err)
+				assert.Equal(t, testCase.expectedError, err.Error())
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, testCase.expectedDirection, result)
+			}
+		})
+	}
+}

pkg/ottl/ottlfuncs/functions.go

@@ -69,6 +69,7 @@ func converters[K any]() []ottl.Factory[K] {
 		NewMinutesFactory[K](),
 		NewMonthFactory[K](),
 		NewNanosecondsFactory[K](),
+		NewNetworkDirectionFactory[K](),
 		NewNowFactory[K](),
 		NewParseCSVFactory[K](),
 		NewParseJSONFactory[K](),