Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
"""
Server Density Plugin
Minecraft player monitor
https://github.com/serverdensity/sd-agent-plugins/
Version: 1.0.0
"""
import json
import logging
import platform
import sys
import subprocess
import time
class Minecraft(object):
"""SD Plugin for reading available statistics from a Minecraft server
over rcon
"""
def __init__(self, agent_config, checks_logger, raw_config):
self.agent_config = agent_config
self.checks_logger = checks_logger
self.raw_config = raw_config
self.version = platform.python_version_tuple()
def run(self):
data = {}
if 'Minecraft' not in self.raw_config:
self.checks_logger.error(
'Plugin config missing for Minecraft server')
return False
host = self.raw_config['Minecraft']['rcon_host']
port = int(self.raw_config['Minecraft']['rcon_port'])
password = self.raw_config['Minecraft']['rcon_password']
try:
import rcon_mc.rcon
import rcon_mc.lib.msocket
except ImportError:
self.checks_logger.error(
'Unable to import "rcon_mc" Python module')
client = rcon_mc.rcon.client(host, port, password)
if not client:
self.checks_logger.error(
'Unable to create connection to server')
return data
try:
response = client.send("/list")
except Exception as exception:
self.checks_logger.error(
'Unable to get response from server. Error: {0}'.format(exception.message))
return data
try:
data['players'] = response.split()[2].split('/')[0]
except Exception as exception:
self.checks_logger.error(
'Unable to get parse reply from server. Error: {0}'.format(exception.message))
return data
if __name__ == '__main__':
"""Standalone test
"""
raw_agent_config = {
'Minecraft': {
'rcon_port': '25575',
'rcon_host': 'localhost',
'rcon_password': 'password'
}
}
main_checks_logger = logging.getLogger('Minecraft')
main_checks_logger.setLevel(logging.DEBUG)
main_checks_logger.addHandler(logging.StreamHandler(sys.stdout))
minecraft_check = Minecraft({}, main_checks_logger, raw_agent_config)
while True:
try:
print json.dumps(minecraft_check.run(), indent=4, sort_keys=True)
except:
main_checks_logger.exception("Unhandled exception")
finally:
time.sleep(60)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.