| Server IP : 153.92.12.50 / Your IP : 216.73.216.91 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/alt/python311/lib/python3.11/site-packages/pyroute2/ |
Upload File : |
import ctypes
import os
import socket
from typing import Literal, Optional, Union
from pyroute2 import netns
from pyroute2.netlink.coredata import CoreConfig, CoreSocketSpec
metric_type = Literal['c', 'g', 'ms']
class StatsDClientSocket(socket.socket):
'''StatsD client.'''
def __init__(
self,
address: Optional[tuple[str, int]] = None,
use_socket: Optional[socket.socket] = None,
flags: int = os.O_CREAT,
libc: Optional[ctypes.CDLL] = None,
):
self.spec = CoreSocketSpec(
CoreConfig(
netns=None, address=address, use_socket=use_socket is not None
)
)
self.status = self.spec.status
self.buffer: str = ''
if use_socket is not None:
fd = use_socket.fileno()
else:
prime = netns.create_socket(
self.spec['netns'], socket.AF_INET, socket.SOCK_DGRAM
)
fd = os.dup(prime.fileno())
prime.close()
super().__init__(fileno=fd)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def commit(self):
self.sendto(self.buffer.encode(), self.spec['address'])
self.buffer = ''
def put(
self, name: str, value: Union[int, str], kind: metric_type
) -> None:
self.buffer += f'{name}:{value}|{kind}\n'
def incr(self, name: str, value: int = 1) -> None:
self.put(name, value, 'c')
self.commit()
def gauge(self, name: str, value: int) -> None:
self.put(name, value, 'g')
self.commit()
def timing(self, name: str, value: int) -> None:
self.put(name, value, 'ms')
self.commit()