MBean interface : ThreadsMonitoringMXBean.java

package threadmon.monitoring;

import com.fasterxml.jackson.core.JsonProcessingException;

/**
 * Interface du MBean pour recuperer les infos des threads 
 * 
 * @author armel.chever
 *
 */
public interface ThreadsMonitoringMXBean {

    /**
     * Retourne un json contenant les infos des threads de la JVM
     * @return  String un json contenant les infos des threads de la JVM
     */
    public String getThreadsStatus() throws JsonProcessingException;
    
}

MBean implémentation:  ThreadsMonitoring.java

package threadmon.monitoring;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.management.ThreadMXBean;

/**
 * Bean JMX permettant de requeter les infos des threads 
 * 
 * @author armel.chever
 *
 */
public class ThreadsMonitoring implements ThreadsMonitoringMXBean {
	
    /**
     * @see simat.common.monitoring.ThreadsMonitoringMXBean#getThreadsStatus()
     */
    public String getThreadsStatus() throws JsonProcessingException {
        
        ThreadMXBean threadMgmt = ((ThreadMXBean) ManagementFactory.getThreadMXBean());
        long[] threadsId = threadMgmt.getAllThreadIds();
        
        ThreadsStatus threadsStatus = new ThreadsStatus();
        
        for (long id : threadsId) {
         
            ThreadInfo info = threadMgmt.getThreadInfo(id);
            
            if (info.getThreadName().contains("exec")) {
	            ThreadState threadState = new ThreadState();
	            threadState.setThreadId(String.valueOf(info.getThreadId()));
	            threadState.setThreadName(info.getThreadName());
	            threadState.setThreadState(String.valueOf(info.getThreadState()));
	            threadState.setAllocatedBytes(threadMgmt.getThreadAllocatedBytes(id));
	            
	            threadsStatus.getThreadsStatus().add(threadState);
            }            
        }
        
      ObjectMapper mapper = new ObjectMapper();
      String status = mapper.writeValueAsString(threadsStatus);
      return status;

    }

}

POJO pour représenter l’état d’un thread : ThreadState.java

package threadmon.monitoring;
/**
 * 
 * Classe representant l'etat d'un Thread a un instant T
 * @author armel.chever
 *
 */
public class ThreadState {

    protected String threadId = "";
    protected String threadName = "";
    protected String threadState = "";
    protected long allocatedBytes = 0;

    /**
     * Construction
     */
    public ThreadState() {
        super();
    }
    
    /**
     * @return the threadId
     */
    public String getThreadId() {
        return threadId;
    }

    /**
     * @param threadId the threadId to set
     */
    public void setThreadId(String threadId) {
        this.threadId = threadId;
    }

    /**
     * @return the threadName
     */
    public String getThreadName() {
        return threadName;
    }

    /**
     * @param threadName the threadName to set
     */
    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    /**
     * @return the threadState
     */
    public String getThreadState() {
        return threadState;
    }

    /**
     * @param threadState the threadState to set
     */
    public void setThreadState(String threadState) {
        this.threadState = threadState;
    }

    /**
     * @return the allocatedBytes
     */
    public long getAllocatedBytes() {
        return allocatedBytes;
    }

    /**
     * @param allocatedBytes the allocatedBytes to set
     */
    public void setAllocatedBytes(long allocatedBytes) {
        this.allocatedBytes = allocatedBytes;
    }
 
}

POJO pour représenter l’état global d’un ensemble de threads : ThreadsStatus.java

package threadmon.monitoring;

import java.util.ArrayList;
import java.util.List;

/**
 * @author armel.chever
 *
 */
public class ThreadsStatus {

	protected List<ThreadState> threadsStatus = new ArrayList<ThreadState>();

	public List<ThreadState> getThreadsStatus() {
		return threadsStatus;
	}

	public void setThreadsStatus(List<ThreadState> threadsStatus) {
		this.threadsStatus = threadsStatus;
	}
}

@Configuration
@EnableAutoConfiguration
@ComponentScan
public class Application {
   	
   public static void main(String[] args) throws Exception {
          	
   	SpringApplication.run(Application.class, args);
          	
   	MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
  	ObjectName name = new ObjectName("monitoring.threadmon.mbean:type=ThreadsMonitoring");
  	ThreadsMonitoring mbean = new ThreadsMonitoring();
  	mbs.registerMBean(mbean, name);
   	
   }
 
}

{  
   "threadsStatus":[  
      {  
         "threadId":"32",
         "threadName":"http-nio-8080-exec-3",
         "threadState":"WAITING",
         "allocatedBytes":118680
      },
      {  
         "threadId":"31",
         "threadName":"http-nio-8080-exec-2",
         "threadState":"WAITING",
         "allocatedBytes":123496
      },
      {  
         "threadId":"30",
         "threadName":"http-nio-8080-exec-1",
         "threadState":"WAITING",
         "allocatedBytes":7063928
      }
   ]
}

01-input-jmx-thread.conf :
input {
   jmx {
     path => "D:\projects\Lab\elastic\logstash-2.3.4\conf\jmxsources"
     polling_frequency => 5
     type => "thread"
     nb_thread => 4
   }
}
 
Puis dans le répertoire « D:\projects\Lab\elastic\logstash-2.3.4\conf\jmxsources » :
jmx_sources.conf :
{
  "host" : "localhost",
  "port" : 11000,
  "queries" : [
	{
  	"object_name" : "monitoring.threadmon.mbean:type=ThreadsMonitoring",
  	"attributes" : [ "ThreadsStatus"],
  	"object_alias" : "Threads"
	}
   ]
}

 
filter {
 
	if [type] == "thread" {
 
           	# Recuperation du json
           		json {
                       		source => "metric_value_string"
                       		target => "jsondata"
           	}
  
           		# Eclatement du flux JSON evenements (1 evenement par thread)
           		split {
                       		field => "[jsondata][threadsStatus]"
           		}
 
           		# Renommage des champs pour la lisibilte
           		mutate {
                       		rename => { "[jsondata][threadsStatus][threadId]" => "thread_id" }
                       		rename => { "[jsondata][threadsStatus][threadName]" => "thread_name" }
                       		rename => { "[jsondata][threadsStatus][threadState]" => "thread_state" }
                       		rename => { "[jsondata][threadsStatus][allocatedBytes]" => "allocated_bytes" }
           		}
           	
           		# Conversion de la memoire en integer
           		mutate {
                       		convert => { "allocated_bytes" => "integer" }    	
           		}
}
}