Coletando LOGs do NetScaler no Elasticsearch e visualizando-os no Grafana

Hoje temos um menu bastante interessante, e se você é responsável por um NetScaler em sua infraestrutura, neste post vamos ver como podemos recolher os seus logs para posteriormente serem processados e finalmente fazer um dashboard em Grafana que mostra em tempo real (ou histórico) Os dados recolhidos.

Então o que eu disse, podemos hacer que nuestro Citrix NetScaler envíe sus logs a Logstash para que los trate y separe en datos estructurados lo que son líneas de distintos logs. Almacenaremos en un índice de Elasticsearch esta información y luego con un conector o ‘Data Source’ de Grafana podremos acceder a estos datos, para visualizarlos en tiempo real o en un periodo de tiempo determinado (últimas 24h, mês…). Y cada uno puede hacerse el dashboard en base al rol que haga su NS, si tiene el servicio Gateway viendo las IP’s públicas que acceden, tráficos… Si tenemos un Content Switching ídem además de destinos… Si tenemos un Load Balancing pues analizar realmente por donde van las conexiones… al final todo lo que se almacena en un log lo podemos explotar y NetScaler es un dispositivo muy interesante para conocer qué pasa dentro, como puedan ser la visualización de las autenticaciones, correctas o incorrectas.

Primeiro, obviamente temos que ter a parte do Elastic Stack instalada, O que foi Armazenamento, Elasticsearch e Kibana; luego ya sería indicarle a nuestro NetScaler que envíe los Logs a Logstash, para o porto de nossa escolha, nos inventaremos uno (desde “Sistema” > “Auditing” > “Syslog Auditing”. Recordar grabar la configuración en el NetScaler.

Lo siguiente será crear un fichero en Logstash que deberá de ingestar, Transformar e enviar dados. Primero escuchar en el puerto UDP indicado en el paso anterior; segundo, processar os logs que recebe e separá-los em campos diferentes para serem finalmente armazenados em nosso Elasticsearch.

Entrada {
        UDP {
                tipo => "Netscaler"
                porta => "1517"
                tags => ["Netscaler"]
        }
}

filtro {

        se [tipo] == "Netscaler" {
                        Grok {
                            correspondência => { "Mensagem" => [
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Context%{SPACE}%{NOME DE UTILIZADOR:Utilizador}@%{IPV4:IP_origen}%{SPACE}-%{SPACE}SessionId:%{SPACE}%{INT:Session_id}%{SPACE}-%{SPACE}User%{SPACE}%{NOME DE UTILIZADOR:Usuario2}%{SPACE}-%{SPACE}Client_ip%{SPACE}%{IPV4:IP_origen2}%{SPACE}-%{SPACE}Nat_ip%{SPACE}\"Mapped Ip\"%{SPACE}-%{SPACE}Vserver%{SPACE}%{IPV4:vServer_ip}:%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}Start_time%{SPACE}\"%{DATE_US:Inicio_fecha}:%{TIME:Inicio_hora}%{SPACE}\"%{SPACE}-%{SPACE}End_time%{SPACE}\"%{DATE_US:Fin_fecha}:%{TIME:Fin_hora}%{SPACE}\"%{SPACE}-%{SPACE}Duration%{SPACE}%{TIME:Duracion}%{SPACE}-%{SPACE}Http_resources_accessed%{SPACE}%{NÚMERO:Http_resources_accessed}%{SPACE}-%{SPACE}NonHttp_services_accessed%{SPACE}%{NÚMERO:NonHttp_services_accessed}%{SPACE}-%{SPACE}Total_TCP_connections%{SPACE}%{NÚMERO:Total_TCP_connections}%{SPACE}-%{SPACE}Total_UDP_flows%{SPACE}%{NÚMERO:otal_UDP_flows}%{SPACE}-%{SPACE}Total_policies_allowed%{SPACE}%{NÚMERO:Total_policies_allowed}%{SPACE}-%{SPACE}Total_policies_denied%{SPACE}%{NÚMERO:Total_policies_denied}%{SPACE}-%{SPACE}Total_bytes_send%{SPACE}%{NÚMERO:Total_bytes_send}%{SPACE}-%{SPACE}Total_bytes_recv%{SPACE}%{NÚMERO:Total_bytes_recv}%{SPACE}-%{SPACE}Total_compressedbytes_send%{SPACE}%{NÚMERO:Total_compressedbytes_send}%{SPACE}-%{SPACE}Total_compressedbytes_recv%{SPACE}%{NÚMERO:Total_compressedbytes_recv}%{SPACE}-%{SPACE}Compression_ratio_send%{SPACE}%{NÚMERO:Compression_ratio_send}\%%{SPACE}-%{SPACE}Compression_ratio_recv%{SPACE}%{NÚMERO:Compression_ratio_recv}\%%{SPACE}-%{SPACE}LogoutMethod%{SPACE}\"%{DATA:LogoutMethod}\"%{SPACE}-%{SPACE}Group\(s\)%{SPACE}\"%{DATA:Group}\"",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Context%{SPACE}%{NOME DE UTILIZADOR:Utilizador}@%{IPV4:IP_origen}%{SPACE}-%{SPACE}SessionId:%{SPACE}%{INT:Session_id}%{SPACE}-%{SPACE}User%{SPACE}%{NOME DE UTILIZADOR:Usuario2}%{SPACE}-%{SPACE}Client_ip%{SPACE}%{IPV4:IP_origen2}%{SPACE}-%{SPACE}Nat_ip%{SPACE}\"Mapped Ip\"%{SPACE}-%{SPACE}Vserver%{SPACE}%{IPV4:vServer_ip}:%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}%{DADOS GANANCIOSOS:Mensagem}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}User%{SPACE}%{NOME DE UTILIZADOR:Utilizador}%{SPACE}-%{SPACE}Client_ip%{SPACE}%{IPV4:IP_origen}%{SPACE}-%{SPACE}%{DADOS GANANCIOSOS:Mensagem}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Source%{SPACE}%{IPV4:IP_origen}:%{NÚMERO:Puerto_origen}%{SPACE}-%{SPACE}Vserver%{SPACE}%{IPV4:vServer_ip}:%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}NatIP%{SPACE}%{IPV4:NAT_ip}:%{NÚMERO:NAT_puerto}%{SPACE}-%{SPACE}Destination%{SPACE}%{IPV4:IP_destino}:%{NÚMERO:Puerto_destino}%{SPACE}-%{SPACE}Delink%{SPACE}Time%{SPACE}%{DATE_US:Delink_fecha}:%{TIME:Delink_hora}%{SPACE}-%{SPACE}Total_bytes_send%{SPACE}%{INT:Total_bytes_enviados}%{SPACE}-%{SPACE}Total_bytes_recv%{SPACE}%{INT:Total_bytes_recibidos}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Source%{SPACE}%{IPV4:IP_origen}:%{NÚMERO:Puerto_origen}%{SPACE}-%{SPACE}Destination%{SPACE}%{IPV4:Destino_ip}:%{NÚMERO:Destino_puerto}%{SPACE}-%{SPACE}Start%{SPACE}Time%{SPACE}%{DATE_US:Inicio_fecha}:%{TIME:Inicio_hora}%{SPACE}-%{SPACE}End%{SPACE}Time%{SPACE}%{DATE_US:Fin_fecha}:%{TIME:Fin_hora}%{SPACE}-%{SPACE}Total_bytes_send%{SPACE}%{INT:Total_bytes_enviados}%{SPACE}-%{SPACE}Total_bytes_recv%{SPACE}%{INT:Total_bytes_recibidos}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Source%{SPACE}%{IPV4:IP_origen}:%{NÚMERO:Puerto_origen}%{SPACE}-%{SPACE}Vserver%{SPACE}%{IPV4:vServer_ip}:%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}NatIP%{SPACE}%{IPV4:NAT_ip}:%{NÚMERO:NAT_puerto}%{SPACE}-%{SPACE}Destination%{SPACE}%{IPV4:IP_destino}:%{NÚMERO:Puerto_destino}%{SPACE}-%{SPACE}Delink%{SPACE}Time%{SPACE}%{DATE_US:Delink_fecha}:%{TIME:Delink_hora}%{SPACE}Total_bytes_send%{SPACE}%{INT:Total_bytes_enviados}%{SPACE}-%{SPACE}Total_bytes_recv%{SPACE}%{INT:Total_bytes_recibidos}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}SPCBId%{SPACE}%{INT:SPCBId}%{SPACE}-%{SPACE}ClientIP%{SPACE}%{IPV4:IP_origen}%{SPACE}-%{SPACE}ClientPort%{SPACE}%{NÚMERO:Puerto_origen}%{SPACE}-%{SPACE}VserverServiceIP%{SPACE}%{IPV4:vServer_ip}%{SPACE}-%{SPACE}VserverServicePort%{SPACE}%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}ClientVersion%{SPACE}%{DATA:Client_version}%{SPACE}-%{SPACE}CipherSuite%{SPACE}\"%{DATA:Cipher_suite}\"(%{SPACE}-%{SPACE}Session%{SPACE}Reuse%{SPACE}-%{SPACE}HandshakeTime%{SPACE}%{INT:Handshake_time}%{SPACE}em)?",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}SPCBId%{SPACE}%{INT:SPCBId}%{SPACE}-%{SPACE}IssuerName%{SPACE}\"%{SPACE}%{DATA:Issuer_name}\"",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}SPCBId%{SPACE}%{INT:SPCBId}%{SPACE}-%{SPACE}SubjectName%{SPACE}\"%{SPACE}%{DATA:Subject_name}\"",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}Context%{SPACE}%{NOME DE UTILIZADOR:Utilizador}@%{IPV4:IP_origen}%{SPACE}-%{SPACE}SessionId:%{SPACE}%{INT:Session_id}%{SPACE}-%{SPACE}%{NOME DO HOST:FQDN}%{SPACE}User%{SPACE}%{NOME DE UTILIZADOR:Usuario2}%{SPACE}:%{SPACE}Group\(s\)%{SPACE}%{DATA:Grupo}%{SPACE}:%{SPACE}Vserver%{SPACE}%{IPV4:vServer_ip}:%{NÚMERO:vServer_puerto}%{SPACE}-%{SPACE}%{DATE_US:Data}:%{TIME:Hora}%{SPACE}:%{SPACE}%{DADOS GANANCIOSOS:Mensagem}",
                              "^<%{DERRUBE:syslog_pri}>%{SPACE}%{DATE_US:Log_fecha}:%{TIME:Log_hora}%{SPACE}%{SYSLOGHOST:NSlog_hostname}%{SPACE}0-PPE-0%{SPACE}:%{SPACE}default%{SPACE}%{PALAVRA:Log_tipo}%{SPACE}%{PALAVRA:Log_evento}%{SPACE}%{INT:Log_id}%{SPACE}0%{SPACE}:%{SPACE}%{DADOS GANANCIOSOS:Mensagem}"
                                       ]
                            }
                        }

                        geoip {
                          source => "IP_origen"
                          target => "geoip"
                          add_field => [ "[geoip][coordinates]", "%{[geoip][Longitude]}" ]
                          add_field => [ "[geoip][coordinates]", "%{[geoip][Latitude]}"  ]
                        }

                        mutate {
                          convert => [ "[geoip][coordinates]", "flutuar" ]
                          convert => [ "Total_bytes_enviados", "Inteiro" ]
                          convert => [ "Total_bytes_recibidos", "Inteiro" ]
                        }
        }
}

saída {

        se ([tipo]=="Netscaler"){

                ElasticSearch {
                   índice => "netscaler-%{+AAAAA. MM.dd}"
                   hosts=> "DIRECCION_IP_ELASTICSEARCH:9200"
                }

        }
}

Así que creamos por ejemplo el siguiente fichero de configuración ‘/etc/logstash/conf.d/netscaler.conf’, yo os dejo mi ejemplo aquí abajo. Mis filtros son muy básicos y de primero de Logstash, pero funcionan 😉 así que si te interesan puedes copiarlos, y si los mejoras me los pasas 🙂 Al final indicaremos en la salida el servidor de Elasticsearch donde mandar los datos y el índice donde se almacenará; recordar que si tenéis autenticación poner los parámetros de ‘username’ & ‘password’ en el output. La verdad que estoy viendo ahora el código y es bastante mejorable, me rayé con los espacios recuerdo… espero me podáis perdonar pero funciona, al menos con las últimas versiones como pueda ser NetScaler 13.0.

Depois que o arquivo de configuração for criado, recordar reiniciar el servicio de Logstash para recargar. Depois, como sempre, iremos a Kibana y una vez los datos estén entrando ya podremos ir a “Gestão” > “Stack management” > “Kibana” > “Padrões de índice” > “Criar padrão de índice” para crear el patrón del índice, Como eu disse, como de costume (neste caso e sem as aspas) ‘netscaler-*’ y tendremos los datos ya en Elasticsearch almacenados de manera correcta. Ahora podríamos conectarnos desde “Descobrir” a nuestro índice de NetScaler y visualizar que está recogiendo datos.

Y luego ya, depois de criar o índice no Kibana, ahora en nuestro Grafana deberíamos crear un “Data Source” que apunte contra nuestro Elasticsearch y el índice de NetScaler. Então é deixar sua imaginação correr solta, faça um Dashboard com diferentes Dashboards, com dados diferentes para visualizar, um Mapa do Mundo con las conexiones entrantes, Um de Estilo Sankey para visualizar IPs de origem/destino e o tráfego que está sendo enviado, em formato de coluna, em impasse para ver datos concretos de por ejemplo los logins correctos, Incorreto, las conexiones…

Todo puede servir para que cojáis ideas y las mejoréis, con esto podréis ver qué pasa en nuestro(s) Citrix NetScaler en tiempo real, Colocando uma atualização automática a cada 10 segundos é muito impressionante, también nos servirá para analizar el resumen de una jornada, o conocer cuando algo extraño sucede… Como de costume, gracias a todos y más a los que movéis este tipo de contenidos en redes sociales 😉