[issue 63] Flink alert sample

[hldnova]

fixes #63

[fpj]

Looks good, @hldnova ! I have only a few requests for change, please address them.

[fpj]

What's this dependency, @hldnova ? What kind of license is this distributed under?

[hldnova]

gson is the google json parser. joda-time, among other things, provides timestamp parsing. They are both under Apache 2.0 license.

[fpj]

If I understand this right, this is expecting some manual input from the user. Could we also have a way to tail access logs coming through files? For example, if I start a web server locally and point logstash to the directory with the access logs, it should be able to tail it, no?

[hldnova]

The common practice is logs -> filebeat -> logstash -> ... Given the number components in the demo, I will make a docker image to set up the end to end environment.

[hldnova]

@fpj, I've updated the sample to read from a file. I also built a docker image to make it easier to set up the environment. I believe i had addressed all the requests from you.

While committing changes, i forgot to set signed-off-by. Had to use amend/rebase to fix them. But somehow the latest commit still complains even if there is a signed-off-by identical to others. I am puzzled and may need some help to fix it.

[fpj]

We add the license header to config files as well, see under config/ in pravega.

[hldnova]

done.

[vijikarthi]

shouldn't we first filter the error log for the response code 500?

[hldnova]

Done. i was thinking about doing something processing together with response code, but decide not to make it complicated.

[vijikarthi]

We are not making use of the log event time to detect the 500 response code for specific window and it looks like it will be based on the processing time. Should we plan to change the code to make use of the log event time, we could modify something like this?

        // initialize the Flink execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        **env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);**

        // create the Pravega source to read a stream of text
        FlinkPravegaReader<String> reader = FlinkPravegaReader.<String>builder()
                .withPravegaConfig(pravegaConfig)
                .forStream(stream)
                .withDeserializationSchema(PravegaSerialization.deserializationFor(String.class))
                .build();

        // add the Pravega reader as the data source
        DataStream<String> inputStream = env.addSource(reader);

        // transform logs
        DataStream<AccessLog> dataStream =
                inputStream.map(new ParseLogData())
                        .filter(a -> a.getStatus().equals("500")).name("filter");
        DataStream<AccessLog> withTimestampsAndWatermarks =
                dataStream.assignTimestampsAndWatermarks(new EventTimeExtractor(Time.seconds(1000)))
                        .name("watermark");

        DataStream<ResponseCount> countStream =
        withTimestampsAndWatermarks
                .keyBy("verb")
                //.timeWindow(Time.seconds(Constants.ALERT_WINDOW))
                .window(TumblingEventTimeWindows.of(Time.seconds(Constants.ALERT_WINDOW)))
                .fold(new ResponseCount(), new FoldFunction<AccessLog, ResponseCount>() {
                    @Override
                    public ResponseCount fold(ResponseCount accumulator, AccessLog value) throws Exception {
                        accumulator.response = value.getVerb();
                        accumulator.count = accumulator.count + 1;
                        return accumulator;
                    }
                }).name("aggregated-count");

        // create an output sink to stdout for verification
        countStream.print().name("print-count");

        // create alert pattern
        Pattern<ResponseCount,?> pattern500 = Pattern.<ResponseCount>begin("500pattern")
            .where(new SimpleCondition<ResponseCount>() {
                @Override
                public boolean filter(ResponseCount value) throws Exception {
                    return value.count >= Constants.ALERT_THRESHOLD;
                }
            });

        PatternStream<ResponseCount> patternStream = CEP.pattern(countStream, pattern500);

        DataStream<Alert> alertStream = patternStream.select(
            new PatternSelectFunction<ResponseCount, Alert>() {
                @Override
                public Alert select(Map<String, List<ResponseCount>> pattern) throws Exception {
                    ResponseCount count = pattern.get("500pattern").get(0);
                    return new Alert(count.response, count.count, "High 500 responses");
                }
            }).name("alert");

        // create an output sink to stdout for verification
        alertStream.print().name("print-alert");


        // execute within the Flink environment
        env.execute("HighCountAlerter");

    public static class EventTimeExtractor extends BoundedOutOfOrdernessTimestampExtractor<AccessLog> {

        public EventTimeExtractor(Time time) { super(time); }

        @Override
        public long extractTimestamp(AccessLog accessLog) {
            return accessLog.getTimestampMillis();
        }
    }

[hldnova]

thanks for the tip. I plan to come up with a few more examples based on the access log. I will use event time in them later. for this example, i'd like to keep it simple and straightforward to follow.

[vijikarthi]

do we really need this filter?

[hldnova]

It was meant to deal with malformed data. No longer relevant. replaced with filter for 500 responses.

[vijikarthi]

looks like we are not using gson instance?

[vijikarthi]

I think Jersey Jackson module (ObjectMapper) can help convert json string to bean if we want to avoid the boiler plate code?

[hldnova]

Done. Use jackson now.

[hldnova]

messed up with merge. submitted replacement: #126