ITIL práctico, El Catálogo de Servicios (parte 6)

Parte 6, Monitorización de las Peticiones de Servicio (BonitaLife + WSO2BAM + Pentaho Suite)

En esta entrada veremos como monitorizar las Peticiones de Servicio, en concreto el tiempo de ejecución de las tareas asociadas y, por supuesto, el tiempo de ejecución global del proceso. Dentro de un ciclo PDCA esta fase correspondería con la fase de ‘Check’ donde analizamos la información de nuestros procesos para verificar que se cumplen los parámetros establecidos y asignamos planes de corrección para las desviaciones detectadas. La arquitectura propuesta es la siguiente:

1. STORE: Mediante los conectores de BonitaLife para WSO2BAM la información referente al tiempo de ejecución de las tareas y procesos serán enviados a los agentes del BAM para su almacenamiento.
2. ANALYZE: Mediante Hive (analytics batch layer) se agregará la información almacenada de los tiempos de ejecución y se exportará a un datawarehouse (en nuestro caso una BD MySQL) para su explotación. Estos scripts se planificarán para su ejecución periódica. En este punto sería recomendable usar un motor optimizado para el trabajo con modelos estrella como, por ejemplo, InfiniDB/MySQL.
3. VISUALIZE/EXPLORE: Mediante WSO2 Gadget Server y Pentaho Suite se explotará la información almacenada en el DW.
   • Ad-Hoc Informes (Pentaho)
   • OLAP Analysis (Pentaho)
   • Dashboards (WSO2 Gadget Server, Community Dashboard Framework)
   • Predictive Models (Weka)

Conectores BonitaLife para WSO2BAM

Mediante los conectores de BonitaLife podemos enviar la información del tiempo de ejecución de las tareas y procesos a los agentes de WSO2 BAM.

Actualmente disponemos de dos conectores:

• blife-wso2bam-process-time, para el tiempo de ejecución de los procesos. Este conector se asocia al evento FINISH del proceso.
• blife-wso2bam-activity-time, para el tiempo de ejecución de las actividades. Este conector se asocia al evento FINISH de la tarea.

En ambos casos los parámetros de configuración son:

donde

• Dirección BAM, dirección donde se encuentra el agente de WSO2 BAM
• Usuario, usuario para la conexión
• Password, password para la conexión
• Nombre del Stream, identificador del stream donde queremos que se almacenen los eventos
• Versión del Stream, versión del stream a publicar.

A medida que los procesos se van ejecutando, la información sobre tiempos de ejecución es enviada a WSO2BAM para su almacenamiento.

Scripts de Análisis HIVE 

Una vez almacenados los eventos relativos al tiempo de ejecución de los procesos la siguiente tarea es preparar los scripts de análisis con el objetivo de agregar la información para su posterior explotación. Estos scripts se planificarán para su ejecución periódica. En nuestro caso, el resultado será la información agregada por día de los tiempos de ejecución y su exportación a una base de datos mysql. Como podremos observar, la estructura de tablas destino será un modelo de estrella que servirá como fuente de análisis para la suite de Pentaho.

dimProcessDefinition

DROP TABLE IF EXISTS blife_wso2bam_process_time;
CREATE EXTERNAL TABLE IF NOT EXISTS blife_wso2bam_process_time (key STRING, version STRING, processDefinitionId BIGINT, processDefinitionName STRING)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler' 
WITH SERDEPROPERTIES ( 
"cassandra.host" = "127.0.0.1", 
"cassandra.port" = "9160",
"cassandra.ks.name" = "EVENT_KS", 
"cassandra.ks.username" = "admin",
"cassandra.ks.password" = "admin", 
"cassandra.cf.name" = "blife_wso2bam_process_time2",
"cassandra.cf.validatorType" = "UTF8Type,UTF8Type,LongType,UTF8Type",
"cassandra.columns.mapping" = ":key, Version, payload_processDefinitionId, payload_processDefinitionName" ); 

CREATE EXTERNAL TABLE IF NOT EXISTS dimProcessDefinition
                         ( processDefinitionId BIGINT, processDefinitionName STRING )
STORED BY 'org.wso2.carbon.hadoop.hive.jdbc.storage.JDBCStorageHandler'
TBLPROPERTIES ( 
'mapred.jdbc.driver.class' = 'com.mysql.jdbc.Driver' ,
'mapred.jdbc.url' = 'jdbc:mysql://localhost/bpmdw' ,
'mapred.jdbc.username' = 'bpmdw' ,
'mapred.jdbc.password' = 'bpmdw' ,
'hive.jdbc.update.on.duplicate' = 'true' ,
'hive.jdbc.primary.key.fields' = 'processDefinitionId' ,
'hive.jdbc.table.create.query' = 'CREATE TABLE dimProcessDefinition 
                                (processDefinitionId BIGINT, processDefinitionName STRING, PRIMARY KEY (processDefinitionId)) ');

insert overwrite table dimProcessDefinition
select distinct cast(processdefinitionid as BIGINT) as ProcessDefinitionId, cast(upper(processDefinitionName) as STRING)
from blife_wso2bam_process_time
where version='1.0.0';

dimActivityDefinition

DROP TABLE IF EXISTS blife_wso2bam_activity_time;
CREATE EXTERNAL TABLE IF NOT EXISTS blife_wso2bam_activity_time (key STRING, version STRING, activityDefinitionId BIGINT, activityDefinitionName STRING)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler' 
WITH SERDEPROPERTIES ( 
"cassandra.host" = "127.0.0.1", 
"cassandra.port" = "9160",
"cassandra.ks.name" = "EVENT_KS", 
"cassandra.ks.username" = "admin",
"cassandra.ks.password" = "admin", 
"cassandra.cf.name" = "blife_wso2bam_activity_time2",
"cassandra.cf.validatorType" = "UTF8Type,UTF8Type,LongType,UTF8Type",
"cassandra.columns.mapping" = ":key, Version, payload_activityDefinitionId, payload_activityDefinitionName" ); 

CREATE EXTERNAL TABLE IF NOT EXISTS dimActivityDefinition
                         ( activityDefinitionId BIGINT, activityDefinitionName STRING )
STORED BY 'org.wso2.carbon.hadoop.hive.jdbc.storage.JDBCStorageHandler'
TBLPROPERTIES ( 
'mapred.jdbc.driver.class' = 'com.mysql.jdbc.Driver' ,
'mapred.jdbc.url' = 'jdbc:mysql://localhost/bpmdw' ,
'mapred.jdbc.username' = 'bpmdw' ,
'mapred.jdbc.password' = 'bpmdw' ,
'hive.jdbc.update.on.duplicate' = 'true' ,
'hive.jdbc.primary.key.fields' = 'activityDefinitionId' ,
'hive.jdbc.table.create.query' = 'CREATE TABLE dimActivityDefinition 
                                (activityDefinitionId BIGINT, activityDefinitionName STRING, PRIMARY KEY (activityDefinitionId)) ');

insert overwrite table dimActivityDefinition
select distinct cast(activitydefinitionid as BIGINT) as ActivityDefinitionId, cast(upper(activityDefinitionName) as STRING)
from blife_wso2bam_activity_time
where version='1.0.0';

dimUserDefinition

DROP TABLE IF EXISTS blife_wso2bam_activity_time;
CREATE EXTERNAL TABLE IF NOT EXISTS blife_wso2bam_activity_time (key STRING, version STRING, userId BIGINT, userName STRING)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler' 
WITH SERDEPROPERTIES ( 
"cassandra.host" = "127.0.0.1", 
"cassandra.port" = "9160",
"cassandra.ks.name" = "EVENT_KS", 
"cassandra.ks.username" = "admin",
"cassandra.ks.password" = "admin", 
"cassandra.cf.name" = "blife_wso2bam_activity_time2",
"cassandra.cf.validatorType" = "UTF8Type,UTF8Type,LongType,UTF8Type",
"cassandra.columns.mapping" = ":key, Version, payload_userId, payload_userName" ); 

CREATE EXTERNAL TABLE IF NOT EXISTS dimUserDefinition
                         ( userDefinitionId BIGINT, userDefinitionName STRING )
STORED BY 'org.wso2.carbon.hadoop.hive.jdbc.storage.JDBCStorageHandler'
TBLPROPERTIES ( 
'mapred.jdbc.driver.class' = 'com.mysql.jdbc.Driver' ,
'mapred.jdbc.url' = 'jdbc:mysql://localhost/bpmdw' ,
'mapred.jdbc.username' = 'bpmdw' ,
'mapred.jdbc.password' = 'bpmdw' ,
'hive.jdbc.update.on.duplicate' = 'true' ,
'hive.jdbc.primary.key.fields' = 'userDefinitionId' ,
'hive.jdbc.table.create.query' = 'CREATE TABLE dimUserDefinition 
                                (userDefinitionId BIGINT, userDefinitionName STRING, PRIMARY KEY (userDefinitionId)) ');

insert overwrite table dimUserDefinition
select distinct cast(userid as BIGINT) as UserDefinitionId, cast(upper(userName) as STRING)
from blife_wso2bam_activity_time
where version='1.0.0';

cubeProcessExecutionTime

DROP TABLE IF EXISTS ProcessExecutionSecs;
CREATE EXTERNAL TABLE IF NOT EXISTS ProcessExecutionSecs (key STRING, version STRING, unixtime BIGINT, processDefinitionId BIGINT, processDefinitionName STRING, segundos BIGINT)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler' 
WITH SERDEPROPERTIES ( 
"cassandra.host" = "127.0.0.1", 
"cassandra.port" = "9160",
"cassandra.ks.name" = "EVENT_KS", 
"cassandra.ks.username" = "admin",
"cassandra.ks.password" = "admin", 
"cassandra.cf.name" = "blife_wso2bam_process_time2",
"cassandra.cf.validatorType" = "UTF8Type,UTF8Type,LongType,LongType,UTF8Type,LongType",
"cassandra.columns.mapping" = ":key, Version, Timestamp, payload_processDefinitionId, payload_processDefinitionName, payload_segundos" ); 

CREATE EXTERNAL TABLE IF NOT EXISTS cubeProcessExecution
                         ( dimDate STRING, processDefinitionId BIGINT, count_process BIGINT, sum_secs BIGINT)
STORED BY 'org.wso2.carbon.hadoop.hive.jdbc.storage.JDBCStorageHandler'
TBLPROPERTIES ( 
'mapred.jdbc.driver.class' = 'com.mysql.jdbc.Driver' ,
'mapred.jdbc.url' = 'jdbc:mysql://localhost/bpmdw' ,
'mapred.jdbc.username' = 'bpmdw' ,
'mapred.jdbc.password' = 'bpmdw' ,
'hive.jdbc.update.on.duplicate' = 'true' ,
'hive.jdbc.primary.key.fields' = 'dimDate,processDefinitionId' ,
'hive.jdbc.table.create.query' = 'CREATE TABLE cubeProcessExecution 
                                (dimDate DATETIME, processDefinitionId BIGINT, count_process BIGINT, sum_secs BIGINT, PRIMARY KEY (dimDate, processDefinitionId)) ');

insert overwrite table cubeProcessExecution
select from_unixtime(cast(unixtime/1000 as BIGINT), 'yyyy-MM-dd') as dimDate, cast(processdefinitionid as BIGINT) as ProcessDefinitionId, count(*) as count_process, sum(segundos) as sum_secs
from ProcessExecutionSecs
where version='1.0.0'
group by from_unixtime(cast(unixtime/1000 as BIGINT), 'yyyy-MM-dd'), cast(processdefinitionid as BIGINT);

cubeActivityExecutionTime

DROP TABLE IF EXISTS ActivityExecutionSecs;
CREATE EXTERNAL TABLE IF NOT EXISTS ActivityExecutionSecs (key STRING, version STRING, unixtime BIGINT, processDefinitionId BIGINT, activityDefinitionId BIGINT, activityDefinitionName STRING, userId BIGINT, segundos BIGINT)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler' 
WITH SERDEPROPERTIES ( 
"cassandra.host" = "127.0.0.1", 
"cassandra.port" = "9160",
"cassandra.ks.name" = "EVENT_KS", 
"cassandra.ks.username" = "admin",
"cassandra.ks.password" = "admin", 
"cassandra.cf.name" = "blife_wso2bam_activity_time2",
"cassandra.cf.validatorType" = "UTF8Type,UTF8Type,LongType,LongType,LongType,UTF8Type,LongType,LongType",
"cassandra.columns.mapping" = ":key, Version, Timestamp, payload_processDefinitionId, payload_activityDefinitionId, payload_activityDefinitionName, payload_userId, payload_segundos" ); 

CREATE EXTERNAL TABLE IF NOT EXISTS cubeActivityExecution
                         ( dimDate STRING, processDefinitionId BIGINT, activityDefinitionId BIGINT, userId BIGINT, count_activity BIGINT, sum_secs BIGINT)
STORED BY 'org.wso2.carbon.hadoop.hive.jdbc.storage.JDBCStorageHandler'
TBLPROPERTIES ( 
'mapred.jdbc.driver.class' = 'com.mysql.jdbc.Driver' ,
'mapred.jdbc.url' = 'jdbc:mysql://localhost/bpmdw' ,
'mapred.jdbc.username' = 'bpmdw' ,
'mapred.jdbc.password' = 'bpmdw' ,
'hive.jdbc.update.on.duplicate' = 'true' ,
'hive.jdbc.primary.key.fields' = 'dimDate,processDefinitionId,activityDefinitionId,userId' ,
'hive.jdbc.table.create.query' = 'CREATE TABLE cubeActivityExecution 
                                (dimDate DATETIME, processDefinitionId BIGINT, activityDefinitionId BIGINT, userId BIGINT, count_activity BIGINT, sum_secs BIGINT, PRIMARY KEY (dimDate, processDefinitionId, activityDefinitionId, userId)) ');

insert overwrite table cubeActivityExecution
select from_unixtime(cast(unixtime/1000 as BIGINT), 'yyyy-MM-dd') as dimDate, cast(processdefinitionid as BIGINT) as ProcessDefinitionId, cast(activitydefinitionid as BIGINT) as ActivityDefinitionId, cast(userid as BIGINT) as UserId, count(*) as count_activity, sum(segundos) as sum_secs
from ActivityExecutionSecs
where version='1.0.0'
group by from_unixtime(cast(unixtime/1000 as BIGINT), 'yyyy-MM-dd'), cast(processdefinitionid as BIGINT), cast(activitydefinitionid as BIGINT),cast(userid as BIGINT);

Exportación de la información agregada

Como podemos observar en las tareas de análisis HIVE, la información generada es exportada a una base de datos MySQL que actúa como intermediario entre las herramientas de visualización/explotación.

Visualización de la información agregada

Una vez tenemos la información agregada la podemos explotar con WSO2 Gadget Server, que viene incluido en la distribución del BAM. Crearemos dos gadgets, uno para el tiempo medio de ejecución de los procesos y otro para el tiempo medio de ejecución de las tareas. El resultado final será el siguiente;

Para la creación de los gadgets utilizaremos la herramienta “Gadget Gen Tool”.

En la siguiente entrada veremos como explotar esta misma información desde la suite de Pentaho. Esta suite nos provee de un toolset de herramientas de Inteligencia de Negocio que nos permite profundizar en nuestros análisis. Entre las herramientas incluidas nos encontramos: Pentaho Report Designer, OLAP (Mondrian), Dashboards (CDF).