Anomaly found in Network Session Traffic (ASIM Network Session schema)

Original query, unchanged.

let min_t = ago(14d);
let max_t = now();
let dt = 1d;
let fieldForDvcAction = "DvcAction";
let fieldForNetworkDirection = "NetworkDirection";
let fieldForNetworkProtocol = "NetworkProtocol";
let AnomalyThreshold = 2.5;
let eps = materialize (_Im_NetworkSession
    | project TimeGenerated
    | where TimeGenerated > ago(5m)
    | count
    | extend Count = Count / 300);
let maxSummarizedTime = toscalar (
    union isfuzzy=true
        (
        NetworkSummary_Protocol_CL
        | where EventTime > min_t
        | summarize max_TimeGenerated=max(EventTime)
        | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
        ),
        (
        NetworkCustomAnalytics_protocol_CL
        | where EventTime_t > min_t
        | summarize max_TimeGenerated=max(EventTime_t)
        | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
        ),
        (
        print(min_t)
        | project max_TimeGenerated = print_0
        )
    | summarize maxTimeGenerated = max(max_TimeGenerated) 
    );
let NetworkSummary_Protocol = materialize(
    union isfuzzy=true 
        (
        NetworkSummary_Protocol_CL
        | where EventTime > ago(1d)
        | project v = int(2)
        ),
        (
        print int(1) 
        | project v = print_0
        )
    | summarize maxv = max(v)
    | extend
        NetworkSummary_Protocol_Exists = (maxv > 1),
        NetworkCustomAnalyticsExists = false
    );
let NetworkCustomAnalytics = materialize(
    union isfuzzy=true 
        (
        NetworkCustomAnalytics_protocol_CL
        | where EventTime_t > ago(1d) 
        | project v = int(2)
        ),
        (
        print int(1) 
        | project v = print_0
        )
    | summarize maxv = max(v)
    | extend
        NetworkCustomAnalyticsExists = (maxv > 1),
        NetworkSummary_Protocol_Exists = false
    );
let allData = union isfuzzy=true 
        (
        (datatable(
    exists: int,
    NetworkSummary_Protocol_Exists: bool,
    NetworkCustomAnalyticsExists: bool
)[
    1, false, false
]
        | where toscalar(eps) > 500
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists)
        | join (
            _Im_NetworkSession(starttime=todatetime(ago(3d)), endtime=now())
            | where TimeGenerated > maxSummarizedTime
            | summarize Count=count()
                by
                NetworkProtocol,
                DstPortNumber,
                DstAppName,
                NetworkDirection,
                DvcAction,
                bin(TimeGenerated, 10m)
            | extend
                EventTime = TimeGenerated,
                Count = toint(Count),
                DstPortNumber = toint(DstPortNumber),
                exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        (datatable(
    exists: int,
    NetworkSummary_Protocol_Exists: bool,
    NetworkCustomAnalyticsExists: bool
)[
    1, false, false
]
        | where toscalar(eps) <= 500
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists
        )
        | join (
            _Im_NetworkSession(starttime=todatetime(ago(4d)), endtime=now())
            | where TimeGenerated > maxSummarizedTime
            | summarize Count=count()
                by
                NetworkProtocol,
                DstPortNumber,
                DstAppName,
                NetworkDirection,
                DvcAction,
                bin(TimeGenerated, 10m)
            | extend
                EventTime = TimeGenerated,
                Count = toint(Count),
                DstPortNumber = toint(DstPortNumber),
                exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        (datatable(exists: int, NetworkSummary_Protocol_Exists: bool)[1, false]
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        )
        | join (
            NetworkCustomAnalytics_protocol_CL
            | where EventTime_t > min_t
            | project
                NetworkProtocol=NetworkProtocol_s,
                DstPortNumber=DstPortNumber_d,
                DstAppName=DstAppName_s,
                NetworkDirection=NetworkDirection_s,
                DvcAction=DvcAction_s,
                Count=count__d,
                EventTime=EventTime_t,
                TimeGenerated,
                Type
            | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber), exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        NetworkSummary_Protocol_CL
        | where EventTime > min_t
        | project
            NetworkProtocol,
            DstPortNumber,
            DstAppName,
            NetworkDirection,
            DvcAction,
            Count=count_,
            EventTime,
            TimeGenerated,
            Type
        | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber)
        )
;
let findVolumneBasedAnomaly = allData
    | make-series total=sum(Count) on EventTime from min_t to max_t step dt
    | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit') 
    | mv-expand anomalies, score, baseline, EventTime, total
    | extend
        anomalies = toint(anomalies),
        score = toint(score),
        baseline = toint(baseline),
        EventTime = todatetime(EventTime),
        total = tolong(total)
    | where EventTime >= ago(1d)
    | where score >= 2 * AnomalyThreshold
;
let findAnomalies = (field: string) {
    allData
    | where isnotempty(column_ifexists(field, ""))
    | make-series total=sum(Count) on EventTime from min_t to max_t step dt by column_ifexists(field, "")
    | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit')
    | mv-expand anomalies, score, baseline, EventTime, total
    | extend
        anomalies = toint(anomalies),
        score = toint(score),
        baseline = toint(baseline),
        EventTime = todatetime(EventTime),
        total = tolong(total)
    | where EventTime >= ago(1d)
    | where score >= 2 * AnomalyThreshold
};
union
    findAnomalies(fieldForDvcAction),
    findAnomalies(fieldForNetworkDirection),
    findAnomalies(fieldForNetworkProtocol),
    findVolumneBasedAnomaly
| extend
    anomalyFieldType  = case (
                        isnotempty(column_ifexists(fieldForDvcAction, "")),
                        "DvcAction",
                        isnotempty(column_ifexists(fieldForNetworkDirection, "")),
                        "NetworkDirection",
                        isnotempty(column_ifexists(fieldForNetworkProtocol, "")),
                        "NetworkProtocol",
                        "TotalVolume"
                    ),
    anomalyFieldValue  = case (
                        isnotempty(column_ifexists(fieldForDvcAction, "")),
                        column_ifexists(fieldForDvcAction, ""),
                        isnotempty(column_ifexists(fieldForNetworkDirection, "")),
                        column_ifexists(fieldForNetworkDirection, ""),
                        isnotempty(column_ifexists(fieldForNetworkProtocol, "")),
                        column_ifexists(fieldForNetworkProtocol, ""),
                        "Overall"
                    )