Skip to content

Publish metrics to NewRelic via OTLP from Statefun running in AWS Managed Flink #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kellinwood wants to merge 1 commit into
base: run-on-aws-managed-flink
Choose a base branch
from
Open

Publish metrics to NewRelic via OTLP from Statefun running in AWS Managed Flink #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion aws-terraform/main.tf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,25 @@ resource "aws_iam_role_policy" "flink_app_metrics_policy" {
})
}

resource "aws_iam_role_policy" "flink_app_secrets_manager_policy" {
name = "SecretsManagerAccessPolicy"
role = aws_iam_role.flink_application_role.id
policy = jsonencode({
Version = "2012-10-17",
Statement = [
{
Action = [
"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret"
],
Effect = "Allow",
Resource = "*"
}
]
})
}


# Reference: https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/kinesisanalyticsv2_application
resource "aws_kinesisanalyticsv2_application" "flink_demo_tf" {
name = "flink-tf-demo-application"
Expand Down Expand Up @@ -203,6 +222,13 @@ resource "aws_kinesisanalyticsv2_application" "flink_demo_tf" {
EVENTS_INGRESS_STREAM_DEFAULT = "${aws_kinesis_stream.flink_demo_ingress.name}"
EVENTS_EGRESS_STREAM_DEFAULT = "${aws_kinesis_stream.flink_demo_egress.name}"
AWS_REGION = data.aws_region.current.name
ENVIRONMENT = "demo"
NAMESPACE = "sandbox-demo"
OTEL_SDK_DISABLED: "false"
OTEL_SERVICE_NAME: "sandbox-statefun"
OTEL_EXPORTER_OTLP_ENDPOINT = "https://otlp.nr-data.net:443"
OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf"
OTEL_EXPORTER_OTLP_HEADERS = "!secret:OTEL_EXPORTER_OTLP_HEADERS"
}
}
}
Expand All @@ -216,7 +242,7 @@ resource "aws_kinesisanalyticsv2_application" "flink_demo_tf" {
monitoring_configuration {
configuration_type = "CUSTOM"
log_level = "INFO"
metrics_level = "TASK"
metrics_level = "APPLICATION"
}
parallelism_configuration {
auto_scaling_enabled = false
Expand Down
28 changes: 25 additions & 3 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,11 @@
<name>Stateful Functions Embedded Java Example</name>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<statefun.version>3.3-1.18</statefun.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<!--- For now, statefun.version=3.3.0.1-1.18 is obtained via a local install of https://github.com/ImagineLearning/flink-statefun/tree/JET-2083-merge-upstream-release-3.3 -->
<statefun.version>3.3.0.1-1.18</statefun.version>
<flink.version>1.18.1</flink.version>
<protobuf.version>3.7.1</protobuf.version>
<protobuf.version>3.25.6</protobuf.version>
<java.version>11</java.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
Expand Down Expand Up @@ -102,6 +103,10 @@
<artifactId>aws-kinesisanalytics-runtime</artifactId>
<version>${kda.runtime.version}</version>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>secretsmanager</artifactId>
</dependency>

<!-- Cloud events SDK -->
<dependency>
Expand Down Expand Up @@ -143,6 +148,23 @@
<version>2.12.1</version>
</dependency>

<!-- OpenTelemetry Metrics -->
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk</artifactId>
<version>1.49.0</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk-metrics</artifactId>
<version>1.49.0</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-exporter-otlp</artifactId>
<version>1.49.0</version>
</dependency>

<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
Expand Down
98 changes: 87 additions & 11 deletions src/main/java/com/example/stateful_functions/Configuration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,37 +4,62 @@
import org.apache.flink.statefun.sdk.kinesis.auth.AwsRegion;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;

import java.lang.reflect.Field;
import java.util.Locale;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.*;

public class Configuration {


private static Logger LOG = LoggerFactory.getLogger(Configuration.class);

public static Properties properties = getProperties();

public static final String ENVIRONMENT = properties.getOrDefault("ENVIRONMENT", "local").toString();
public static String NAMESPACE = properties.getOrDefault("NAMESPACE", "jetstream-local").toString();

public static String AWS_REGION = properties.getOrDefault("AWS_REGION", "us-east-1").toString();
public static boolean OTEL_SDK_DISABLED = properties.getOrDefault("OTEL_SDK_DISABLED", "true").equals("true");
public static String OTEL_SERVICE_NAME = properties.getOrDefault("OTEL_SERVICE_NAME", "jetstream-statefun").toString();
public static String OTEL_EXPORTER_OTLP_ENDPOINT = properties.getOrDefault("OTEL_EXPORTER_OTLP_ENDPOINT", "").toString();
public static String OTEL_EXPORTER_OTLP_HEADERS = properties.getOrDefault("OTEL_EXPORTER_OTLP_HEADERS", "").toString();
public static String OTEL_EXPORTER_OTLP_PROTOCOL = properties.getOrDefault("OTEL_EXPORTER_OTLP_PROTOCOL", "").toString();
public static String OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE =
properties.getOrDefault("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE","Delta").toString();
public static long OTEL_METRIC_EXPORT_INTERVAL =
Long.parseLong(properties.getOrDefault("OTEL_METRIC_EXPORT_INTERVAL", "60000").toString());
public static long OTEL_METRIC_EXPORT_TIMEOUT =
Long.parseLong(properties.getOrDefault("OTEL_METRIC_EXPORT_TIMEOUT", "30000").toString());


public static String INGRESS_KINESIS_STREAM_NAME = properties.getOrDefault("EVENTS_INGRESS_STREAM_DEFAULT", "example-ingress-stream").toString();
public static String EGRESS_KINESIS_STREAM_NAME = properties.getOrDefault("EVENTS_EGRESS_STREAM_DEFAULT", "example-egress-stream").toString();

public static boolean IS_LOCAL_DEV = properties.getOrDefault("IS_LOCAL_DEV", "false").equals("true");


public static boolean USE_ENHANCED_FANOUT = properties.getOrDefault("USE_ENHANCED_FANOUT", "true").equals("true");
public static String ENHANCED_FANOUT_NAME = properties.getOrDefault("ENHANCED_FANOUT_NAME", "example-enhanced-fanout").toString();

public static String APP_VERSION = properties.getOrDefault("app.version", "0.1").toString();


public static final AwsRegion getAwsRegion() {
return getAwsRegion(properties);
}

if (properties.containsKey("AWS_REGION") && !properties.containsKey("AWS_ENDPOINT")){
return AwsRegion.ofId(properties.get("AWS_REGION").toString());
private static AwsRegion getAwsRegion(Properties _properties) {

if (_properties.containsKey("AWS_REGION") && !_properties.containsKey("AWS_ENDPOINT")){
return AwsRegion.ofId(_properties.get("AWS_REGION").toString());
} else {

AwsRegion region = AwsRegion.ofCustomEndpoint(
properties.getOrDefault("AWS_ENDPOINT", "https://localhost:4566").toString(),
properties.getOrDefault("AWS_REGION", "us-east-1").toString()
_properties.getOrDefault("AWS_ENDPOINT", "https://localhost:4566").toString(),
_properties.getOrDefault("AWS_REGION", "us-east-1").toString()
);

try {
Expand All @@ -45,10 +70,10 @@ public static final AwsRegion getAwsRegion() {
{
field.set(region, "http://host.docker.internal:4566");
}
} catch (Exception ex) {
}
catch (Exception ex) {
}
return region;

}
}

Expand Down Expand Up @@ -76,7 +101,58 @@ private static Properties getProperties() {
catch (Exception x) {
LOG.warn(x.getMessage(), x);
}
return properties;

return resolveSecretValueReferences(properties);
}

// Replace property values of the form "!secret:<secret-name>" with the actual secret value from AWS Secrets Manager
private static Properties resolveSecretValueReferences(Properties properties) {

// Find property key/value pairs where the value has a secret reference
Map<String,String> propertiesWithSecretsResolved = new HashMap<>();
properties.stringPropertyNames().forEach(propertyName -> {
String propertyValue = properties.getProperty(propertyName);
if (propertyValue.startsWith("!secret:")) {
String secretName = propertyValue.substring(8);
propertiesWithSecretsResolved.put(propertyName, secretName);
}
});

// Nothing to do if there are no secrets to resolve
if (propertiesWithSecretsResolved.isEmpty()) {
return properties;
}

String awsRegionId = properties.get("AWS_REGION").toString();
LOG.info("Resolving configuration secrets using AWS_REGION={}", awsRegionId);

// Create a Secrets Manager client
try (SecretsManagerClient client = SecretsManagerClient.builder()
.region(Region.of(awsRegionId))
.build()) {

// For each property with a secret reference, get the secret value from AWS Secrets Manager and replace
// the value in the properties map
propertiesWithSecretsResolved.entrySet().forEach(entry -> {
String propertyName = entry.getKey();
String secretName = entry.getKey();

GetSecretValueRequest getSecretValueRequest = GetSecretValueRequest.builder()
.secretId(secretName)
.build();

try {
properties.put(propertyName, client.getSecretValue(getSecretValueRequest).secretString());
}
catch (Exception e) {
LOG.warn("Failed to get secret value for {}", secretName, e);
}
});
}
catch (Exception e) {
LOG.warn("Failed init SecretsManagerClient: ", e);
}

return properties;
}
}
5 changes: 5 additions & 0 deletions src/main/java/com/example/stateful_functions/Main.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,11 @@ public static void main(String... args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StatefulFunctionsConfig stateFunConfig = StatefulFunctionsConfig.fromEnvironment(env);

stateFunConfig.setMetricFunctionNamespaceKey("ns");
stateFunConfig.setMetricFunctionTypeKey("type");
stateFunConfig.setMetricGroupAdapterFactoryClassName("com.example.stateful_functions.metrics.JetstreamMetricGroupAdapterFactory");

stateFunConfig.setProvider((StatefulFunctionsUniverseProvider) (classLoader, statefulFunctionsConfig) -> {
Modules modules = Modules.loadFromClassPath(stateFunConfig);
return modules.createStatefulFunctionsUniverse();
Expand Down
3 changes: 3 additions & 0 deletions src/main/java/com/example/stateful_functions/function/product/ProductStatefulFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,18 +101,21 @@ private void handleFunctionSubscriptionEvent(Context context, CloudEvent subscri
FunctionSubscriptionDetails subscriptionDetails = cloudEventDataAccess.toFunctionSubscriptionDetails(subscriptionEvent);
FunctionSubscriber subscriber = FunctionSubscriberUtil.subscriberFromSubscription(subscriptionDetails);
if (subscriptionDetails.getAction() == FunctionSubscriptionAction.UNSUBSCRIBE) {
context.metrics().counter("unsubscribed").inc();
subscribers.remove(subscriber.getSubscriberId());
return;
}

if (subscriptionDetails.getAction() == FunctionSubscriptionAction.SUBSCRIBE) {
context.metrics().counter("subscribed").inc();
subscribers.set(subscriber.getSubscriberId(), subscriber);
}

notifySubscriber(context, subscriber, createEventFromState());
}

private void notifySubscriber(Context context, FunctionSubscriber subscriber, CloudEvent productEvent) {
context.metrics().counter("notify_subscriber").inc();
send(context, new FunctionType(subscriber.getNamespace(), subscriber.getType()), subscriber.getId(), productEvent);
}

Expand Down
49 changes: 49 additions & 0 deletions src/main/java/com/example/stateful_functions/metrics/JetstreamMetricCounter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
package com.example.stateful_functions.metrics;

import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.metrics.LongGauge;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.SimpleCounter;

public class JetstreamMetricCounter implements Counter {

private final Counter flinkCounter = new SimpleCounter();

private final LongGauge otelGauge;
private final Attributes otelAttributes;

public JetstreamMetricCounter(LongGauge otelGauge, Attributes otelAttributes) {
this.otelGauge = otelGauge;
this.otelAttributes = otelAttributes;
}

@Override
public void inc() {
flinkCounter.inc();
otelGauge.set(flinkCounter.getCount(), otelAttributes);
}

@Override
public void inc(long l) {
otelGauge.set(flinkCounter.getCount() + l, otelAttributes);
flinkCounter.inc(l);

}

@Override
public void dec() {
flinkCounter.dec();
otelGauge.set(flinkCounter.getCount(), otelAttributes);
}

@Override
public void dec(long l) {
otelGauge.set(flinkCounter.getCount() - l, otelAttributes);
flinkCounter.dec(l);
}

@Override
public long getCount() {
return flinkCounter.getCount();
}
}
Loading