custom search command mvrowexpand mv expand jt

commands.conf

[mvrowexpand]
filename = mvrowexpand.py

comment.py

import cherrypy
import datetime
import json
import splunk.appserver.mrsparkle.controllers as controllers
import splunklib.client as client
from splunk.appserver.mrsparkle.lib.decorators import expose_page
from splunk.appserver.mrsparkle.lib.routes import route

service = client.connect(
    host="localhost",
    port=8089,
    username="hobbes3",
    password="hobbes3"
)

class Comment(controllers.BaseController):
    @route("/")
    @expose_page(must_login=False, methods=["POST"])
    def comment(self, **kwargs):
        try:
            global service

            #session_key =  cherrypy.session.get("sessionKey")

            #users = service.users
            #roles = []

            #for user in users:
            #    if user.name = kwargs["user"]:
            #        commenter = user

            #for role in commenter.role_entities:
            #    roles.append(role.name)

            output = kwargs

            output["time"] = datetime.datetime.now().strftime("%s.%f")[:-3]

            comments_index = service.indexes["selftrades_comments"]
            comments_index.submit(json.dumps(output, separators=(",", ":")), sourcetype="comment")

            return self.render_json(output)
        except Exception as e:
            cherrypy.response.status = 500
            output = {
                "error": repr(e)
            }
            return self.render_json(output)

comments.css

.bubble .header {
    font-size: 80%;
    color: grey;
}

.chat {
    width: 600px;
    max-height: 300px;
    overflow: auto;
}

.bubble{
    background-color: #F2F2F2;
    border-radius: 5px;
    box-shadow: 0 0 6px #B2B2B2;
    display: inline-block;
    padding: 10px 18px;
    position: relative;
    vertical-align: top;
    clear: both;
}

.bubble::before {
    background-color: #F2F2F2;
    content: "\00a0";
    display: block;
    height: 16px;
    position: absolute;
    top: 11px;
    transform:             rotate( 29deg ) skew( -35deg );
        -moz-transform:    rotate( 29deg ) skew( -35deg );
        -ms-transform:     rotate( 29deg ) skew( -35deg );
        -o-transform:      rotate( 29deg ) skew( -35deg );
        -webkit-transform: rotate( 29deg ) skew( -35deg );
    width:  20px;
}

.bubble pre {
    background-color: #F2F2F2;
    border: none;
}

.them {
    float: left;
    margin: 5px 45px 5px 20px;
}

.them::before {
    box-shadow: -2px 2px 2px 0 rgba( 178, 178, 178, .4 );
    left: -9px;
}

.you {
    float: right;
    margin: 5px 40px 5px 45px;
}

.you::before {
    box-shadow: 2px -2px 2px 0 rgba( 178, 178, 178, .4 );
    right: -9px;
}

comments.js

require([
    'splunkjs/mvc/tableview',
    'splunkjs/mvc',
    'jquery',
    'underscore',
    'splunk.config',
    '../app/selftrades/moment.min',
    'splunkjs/mvc/simplexml/ready!'
],
function(
    TableView,
    mvc,
    $,
    _,
    SplunkConfig,
    moment
) {
    var INDEX = "selftrades_comments";

    var commenter = SplunkConfig['USERNAME'];
    var service = mvc.createService();
    var indexes = service.indexes();

    var CommentsBaseRowExpansionRenderer = TableView.BaseRowExpansionRenderer.extend({
        initialize: function(id) {
            var that = this;

            that._id = id;

            indexes.fetch(function(err, myindexes) {
                that._index = indexes.item(INDEX);
            });
        },
        canRender: function(row_data) {
            return true;
        },
        render: function($container, row_data) {
            function html_encode(value) {
                return $('<div/>').text(value).html();
            }

            var that = this;

            var id = row_data.cells[row_data.fields.indexOf(that._id)].value;

            var comments = $('<div class="chat"><div id="comment_status" class="bubble them"><pre>Loading comments...</pre></div></div>');
            var textarea = $('<textarea rows="3" cols="100" style="width: auto;" placeholder="Add your comment here. Commenting as ' + commenter + '."></textarea>');
            var submit_button = $('<button>Add comment</button>');

            var search = "search index=" + INDEX + " sourcetype=comment " + that._id + "=" + id + " | table time user comment | reverse";
            service.oneshotSearch(search, {}, function(err, results) {
                var fields = results.fields;
                var rows = results.rows;

                if(rows.length === 0) {
                    $("#comment_status pre", comments).text("No comments.");
                }
                else {
                    $("#comment_status", comments).remove();
                }

                for(var i = 0; i < rows.length; i++) {
                    var row = rows[i];

                    var epoch = row[fields.indexOf("time")];
                    var time = moment(epoch, "X").calendar();
                    var formatted_time = moment(epoch, "X").format();

                    var user = row[fields.indexOf("user")];
                    var comment = row[fields.indexOf("comment")];

                    var bubble = $('<div class="bubble" title="' + formatted_time + '"></div>');
                    var bubble_content = '';

                    if(user === commenter) {
                        bubble_content += '<div class="header">' + time + '</div>';
                        bubble.addClass("you");
                    }
                    else {
                        bubble_content += '<div class="header">' + user + ' ' + time + '</div>';
                        bubble.addClass("them");
                    }

                    bubble_content += '<pre>' + html_encode(comment) + '</pre>';

                    bubble.append(bubble_content);
                    comments.append(bubble);
                }

                comments.animate({ scrollTop: comments.prop("scrollHeight")}, 1000);
            });

            $container.append(comments);
            $container.append("<br/>");
            $container.append(textarea);
            $container.append("<br/>");
            $container.append(submit_button);

            submit_button.on("click", function() {
                submit_button.prop("disabled", true);

                var comment = textarea.val();
                var comment_length = comment.length;

                if(comment_length > 1000) {
                    alert("Please keep comments under 1000 characters. Your comment is " + comment_length + " characters long.");
                    submit_button.prop("disabled", false);
                }
                else {
                    var data = {
                        "id": id,
                        "user": commenter,
                        "comment": comment
                    };

                    $.post("/en-US/custom/selftrades/comment", data, function(response) {
                        textarea.val("");
                        submit_button.prop("disabled", false);

                        if(response.error) {
                            alert(response.error);
                        }
                        else {
                            var epoch = response.time;
                            var time = moment(epoch, "X").calendar();
                            var formatted_time = moment(epoch, "X").format();

                            var bubble = $('<div class="bubble you" title="' + formatted_time + '"></div>');
                            var bubble_content = '<div class="header">' + time + '</div><pre>' + html_encode(comment) + '</pre>';

                            $("#comment_status", comments).remove();
                            bubble.append(bubble_content);
                            comments.append(bubble);

                            comments.animate({ scrollTop: comments.prop("scrollHeight")}, 1000);
                        }
                    });
                }
            });
        }
    });

    var comment_table = mvc.Components.getInstance("comment_table");
    comment_table.getVisualization(function(tableView) {
        tableView.addRowExpansionRenderer(new CommentsBaseRowExpansionRenderer("id"));
    });
});

mvrowexpand.py

#!/usr/bin/python

import logging
import sys
import splunk.Intersplunk
import pdb
import pprint

def main():
    args, kwargs = splunk.Intersplunk.getKeywordsAndOptions()

    logging.debug("args: %s, kwargs: %s", args, kwargs)

    if len(args) < 1:
        splunk.Intersplunk.parseError("Missing argument.")

    rows = splunk.Intersplunk.readResults(None, None, True)

    substring = args[0]
    results = []

    #rows = [
    #    {"A": "apple", "B": "1 2 3", "C": "cat"}
    #]

    for i, row in enumerate(rows):
        logging.debug("i: %s", i)
        logging.debug("row: %s", row)

        for key, value in row.iteritems():
            logging.debug("key: %s", key)
            logging.debug("value: %s", value)
            if substring in key:
                mv_value = value.split()
                logging.debug("mv_value: %s", mv_value)
                mv_len = len(mv_value)
                break

        logging.debug("mv_len: %s", mv_len)

        for k in range(0, mv_len):
            logging.debug("k: %s", k)
            new_row = {}

            for key, value in row.iteritems():
                logging.debug("key: %s", key)
                logging.debug("value: %s", value)

                if substring in key:
                    mv_value = value.split()
                    logging.debug("mv_value: %s", mv_value)
                    new_row[key] = mv_value[k]
                else:
                    new_row[key] = value

            logging.debug("new_row: %s", new_row)
            results.append(new_row)

    splunk.Intersplunk.outputResults(results)

logging.basicConfig(
    filename='/tmp/mvrowexpand.log',
    format='%(asctime)s %(levelname)s %(message)s',
    #level=logging.DEBUG
    level=logging.INFO
)

logging.info("\n\n\n---- STARTING mvrowexpand.py ----")

try:
     main()
except Exception, e:
    logging.exception("Unhanded top-level exception")
    splunk.Intersplunk.parseError("Exception! %s (See /tmp/mvrowexpand.log)" % (e,))

selftrades_alert_high_count_per_team

index=selftrades_summary search_name=selftrades_summary
| stats dc(audit_trail_id) as selftrade_dc_audit_trail_id by selftrade_team _time
| where selftrade_dc_audit_trail_id>100
| eval id=sha1(random().time().selftrade_team)

selftrades_alert_high_st_vol_%_per_audit_trail_id

index=selftrades_summary search_name=selftrades_summary
| eval quantity=if(selftrade_MsgType=6, abs(selftrade_QUANTITY), null())
| stats min(selftrade_vol) as selftrade_vol min(quantity) as selftrade_st_vol by _time audit_trail_id selftrade_team selftrade_JumpSymbolName
| rename audit_trail_id as selftrade_audit_trail_id
| eval selftrade_st_vol_%=selftrade_st_vol/selftrade_vol*100
| where 'selftrade_st_vol_%'>0.1 AND selftrade_vol>1000
| eval id=sha1(random().time().selftrade_audit_trail_id)

selftrades_alert_ill_formed_alerts

index=selftrades_summary search_name!=selftrades_summary
| stats count as selftrade_count dc(id) as selftrade_dc_id by search_name
| rename search_name as selftrade_offending_search_name
| eval selftrade_reason=null()
| eval selftrade_reason=mvappend(selftrade_reason, if(selftrade_dc_id=0, "No id field set.", null()))
| eval selftrade_reason=mvappend(selftrade_reason, if(!match(selftrade_offending_search_name, "^selftrades_alert_"), "Search name doesn't start with selftrades_alert_.", null()))
| where isnotnull(selftrade_reason)
| eval id=sha1(random().time().selftrade_offending_search_name)

selftrades_alert_low_events_in_summary

index=selftrades_summary search_name=selftrades_summary
| stats count as selftrade_count
| where selftrade_count<1000
| addinfo
| eval _time=strptime(strftime(info_max_time, "%F"), "%F")
| fields - info_*
| eval id=sha1(random().time().selftrade_count)

selftrades_summary

`selftrades_events`
[
  search `selftrades_events` MsgType=6
  | eval side=if(exchange="ose",
    if(Quantity>0, "buy", "sell"),
    if(match(TradeAttrPretty, "BUY"), "buy", "sell")
  )
  | join type=left OrderId [
    search index=jumpapplications sourcetype=DropGW-RJ-Transaction exchange=ose SECONDARY_EXCHANGE_TRANSACTION_ID=*
    | stats count by OrderId SECONDARY_EXCHANGE_TRANSACTION_ID
    | fields - count
  ]
  | rex field=EXCHANGE_TRANSACTION_ID "(?<cme_exchange_transaction_id>.{9})$"
  | eval exchange_transaction_id=case(
    exchange="cme", cme_exchange_transaction_id,
    exchange="ose", SECONDARY_EXCHANGE_TRANSACTION_ID,
    0=0, EXCHANGE_TRANSACTION_ID
  )
  | stats dc(side) list(OrderId) by exchange_transaction_id exchange JumpSymbolName FillPrice team 
  | rename list(*) as * dc(*) as dc_*
  | where dc_side>1
  | table OrderId
]
| join max=0 OrderId
[
  search `selftrades_events` MsgType=6
  | eval side=if(exchange="ose",
    if(Quantity>0, "buy", "sell"),
    if(match(TradeAttrPretty, "BUY"), "buy", "sell")
  )
  | join type=left OrderId [
    search index=jumpapplications sourcetype=DropGW-RJ-Transaction exchange=ose SECONDARY_EXCHANGE_TRANSACTION_ID=*
    | stats count by OrderId SECONDARY_EXCHANGE_TRANSACTION_ID
    | fields - count
  ]
  | rex field=EXCHANGE_TRANSACTION_ID "(?<cme_exchange_transaction_id>.{9})$"
  | eval exchange_transaction_id=case(
    exchange="cme", cme_exchange_transaction_id,
    exchange="ose", SECONDARY_EXCHANGE_TRANSACTION_ID,
    0=0, EXCHANGE_TRANSACTION_ID
  )
  | stats dc(side) list(OrderId) by exchange_transaction_id exchange JumpSymbolName FillPrice team 
  | rename list(*) as * dc(*) as dc_*
  | where dc_side>1
  | eval audit_trail_id=exchange_transaction_id.":".exchange.":".JumpSymbolName.":".FillPrice.":".team
  | table OrderId audit_trail_id
  | mvexpand OrderId
]
| eventstats values(Type) as Types by OrderId
| eval order_type=if(Types="NITRO", "NITRO", "RAMJET")
| eval TXN_msg=if(MsgType=0 AND order_type="NITRO", "ORDER_SUBMIT_REQUEST", TXN_msg)
| eval time=
  if(ORDER_CANCEL_REASON="SELF_TRADE_PROTECTION",
    if(match(TXN_msg, "ORDER_(CANCEL|SUBMIT)_RESPONSE"), EXCHANGE_REQUEST_TIME, CommitTime),
    if(TXN_msg="ORDER_CANCEL_RESPONSE", GATEWAY_REQUEST_TIME, CommitTime)    
  )
| sort 0 -time
| eval strategy=Account.":".JumpSymbolName.":".Trader
| eval algorithm=team.":".ALGORITHM_ID
| eval time_=_time
| lookup riskStrategyNames.csv Trader Account JumpSymbolName output RiskStrategy LegalEntity
| lookup MarketStructureSymbols.csv JumpId output InstrumentType SubExchange
| table audit_trail_id Account ALGORITHM_ID CommitTime EXCHANGE_TRANSACTION_ID exchange EXCHANGE_EVENT_TIME EXCHANGE_ORDER_ID EXCHANGE_REQUEST_TIME EXCHANGE_RESPONSE_TIME FillPrice gateway GATEWAY_REQUEST_TIME InstrumentType JumpSymbolName LegalEntity msg MsgType ORDER_ATTRIBUTES ORDER_CANCEL_REASON OrderAttrPretty OrderId PRICE QUANTITY RiskStrategy SENDER_ID Session STRATEGY_REQUEST_TIME SubExchange team time time_ TradeAttrPretty TradeDate Trader TXN_msg Type UNIQUE_TRANSACTION_ID
| join type=left JumpSymbolName
[
  search `selftrades_events` MsgType=6
  | eval quantity=abs(Quantity)
  | stats sum(quantity) as vol by JumpSymbolName
]
| rex field=EXCHANGE_TRANSACTION_ID "(?<cme_exchange_transaction_id>.{9})$"
| join type=left OrderId [
  search index=jumpapplications sourcetype=DropGW-RJ-Transaction exchange=ose SECONDARY_EXCHANGE_TRANSACTION_ID=*
  | stats count by OrderId SECONDARY_EXCHANGE_TRANSACTION_ID
  | fields - count
]
| eval exchange_transaction_id=case(
  exchange="cme", cme_exchange_transaction_id,
  exchange="ose", SECONDARY_EXCHANGE_TRANSACTION_ID,
  0=0, EXCHANGE_TRANSACTION_ID
)
| rex field=audit_trail_id "^(?<exchange_transaction_id_true>[^:]+):[^:]+:(?<jump_symbol_name_true>[^:]+)"
| where exchange_transaction_id_true=exchange_transaction_id OR isnull(exchange_transaction_id) AND JumpSymbolName=jump_symbol_name_true OR isnull(JumpSymbolName)
| eventstats values(SubExchange) as SubExchange_full by audit_trail_id
| eval SubExchange_full=coalesce(SubExchange_full, exchange)
| fields - exchange_transaction_id* jump_symbol_name_true
| fillnull value="null"
| foreach * [eval <<FIELD>>=if(<<FIELD>>="", "null", <<FIELD>>)]
| eval LegalEntity=replace(LegalEntity, " ", "_")
| eval strategy=Account+JumpSymbolName+Trader
| eval algorithm=team+ALGORITHM_ID
| stats list(*) as selftrade_* dc(strategy) dc(algorithm) by audit_trail_id
| rename dc(*) as dc_*
| eval FLAG_LEG_OF_SPREAD=if(match(selftrade_TradeAttrPretty, "(?<!MULTI_)LEG"), 1, 0)
| eval FLAG_IS_NITRO=if(selftrade_Type="NITRO", 1, 0)
| eval FLAG_DIFFERENT_STRATEGIES=if(dc_strategy>1, 1, 0)
| eval FLAG_DIFFERENT_LEGAL_ENTITY=if(mvcount(selftrade_LegalEntity)>1, 1, 0)
| eval FLAG_DIFFERENT_SESSIONS=if(mvcount(selftrade_Session)>1, 1, 0)
| eval FLAG_DIFFERENT_SENDER_IDS=if(mvcount(selftrade_SENDER_ID)>1, 1, 0)
| eval FLAG_DIFFERENT_ALGORITHM_IDS=if(selftrade_exchange="eurex" OR selftrade_exchange="xetra", if(dc_algorithm>1, 1, 0), -1)
| eval FLAG_NO_ORDER_SUBMIT_REQUESTS=if(selftrade_TXN_msg="ORDER_SUBMIT_REQUEST", 0, 1)
| eval FLAG_NO_ORDER_MESSAGES=if(selftrade_MsgType=6 AND mvcount(selftrade_MsgType)=1, 1, 0)
| eval FLAG_NO_ATTEMPT_TO_CANCEL=if(selftrade_TXN_msg="ORDER_CANCEL_REQUEST", 0, if(selftrade_Type="NITRO" AND selftrade_TXN_msg="ORDER_CANCEL_RESPONSE", 0, 1))
| eval FLAG_INFLIGHT=if(
  selftrade_TXN_msg="ORDER_CANCEL_REQUEST",
  if(mvfind(selftrade_TXN_msg, "ORDER_CANCEL_REQUEST")>mvfind(selftrade_TXN_msg, "ORDER_SUBMIT_REQUEST"), 1, 0),
  if(mvfind(selftrade_TXN_msg, "ORDER_CANCEL_RESPONSE")>mvfind(selftrade_TXN_msg, "ORDER_SUBMIT_RESPONSE"), 1, 0)
)
| eval lc_request_orderid=mvindex(selftrade_OrderId, mvfind(selftrade_TXN_msg, "ORDER_CANCEL_REQUEST"))
| eval lc_response_orderid=mvindex(selftrade_OrderId, mvfind(selftrade_TXN_msg, "ORDER_CANCEL_RESPONSE"))
| eval lc_submit_orderid=mvindex(selftrade_OrderId, mvfind(selftrade_TXN_msg, "ORDER_SUBMIT_REQUEST"))
| eval FLAG_LATE_CANCEL=if(
  TXN_msg="ORDER_CANCEL_REQUEST",
  if(mvfind(selftrade_TXN_msg, "ORDER_CANCEL_REQUEST")<mvfind(selftrade_TXN_msg, "ORDER_SUBMIT_REQUEST") AND lc_request_orderid!=lc_submit_orderid, 1, 0),
  if(mvfind(selftrade_TXN_msg, "ORDER_CANCEL_RESPONSE")<mvfind(selftrade_TXN_msg, "ORDER_SUBMIT_REQUEST") AND lc_response_orderid!=lc_submit_orderid, 1, 0)
)
| fields - lc_* dc_*
| eval FLAG_SELF_TRADE_PROTECTION=if(selftrade_ORDER_CANCEL_REASON="SELF_TRADE_PROTECTION", 1, 0)
| mvrowexpand selftrade_
| foreach selftrade_* [eval <<FIELD>>=if(<<FIELD>>="null", null(), <<FIELD>>)]
| table *
| addinfo
| eval _time=strptime(strftime(info_max_time, "%F"), "%F")
| fields - info_*

web.conf

[endpoint:comment]