Merge pull request #2 from sourceplusplus/live-meter

BFergerson · web-flow · commit a35c3f2f0e2a · 2021-11-17T17:08:15.000-05:00
Live meter
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
@@ -26,7 +26,7 @@ jobs:
           PROBE_VERSION="$(python setup.py --version)"
           echo "SPP_PROBE_VERSION=$PROBE_VERSION" >> $GITHUB_ENV
       - run: python setup.py sdist
-      - run: cp dist/sourceplusplus-${{ env.SPP_PROBE_VERSION }}.tar.gz e2e && ls && pwd
+      - run: cp dist/sourceplusplus-${{ env.SPP_PROBE_VERSION }}.tar.gz e2e
       - run: cd e2e && docker-compose up -d
       - name: Docker IPs
         run: docker inspect --format='{{.Name}} - {{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $(docker ps -aq)
@@ -47,7 +47,7 @@ jobs:
 
       - name: Download & install spp-cli
         run: |
-          curl -O -J -L https://github.com/sourceplusplus/interface-cli/releases/latest/download/spp-cli && chmod +x spp-cli && mv spp-cli /usr/local/bin/spp-cli
+          curl -O -J -L https://github.com/sourceplusplus/interface-cli/releases/latest/download/spp-cli-linux64.zip && unzip spp-cli-linux64.zip && chmod +x spp-cli && mv spp-cli /usr/local/bin/spp-cli
 
       - name: Live breakpoint (false condition)
         run: |
diff --git a/e2e/docker-compose.yml b/e2e/docker-compose.yml
@@ -46,15 +46,10 @@ services:
     ports:
       - "11800:11800"
     depends_on:
-      - es-server
       - spp-platform
     links:
-      - es-server
       - spp-platform
-    restart: on-failure
     environment:
-      - SW_STORAGE=elasticsearch
-      - SW_STORAGE_ES_CLUSTER_NODES=es-server:9200
       - SPP_PLATFORM_HOST=spp-platform
       - SPP_PLATFORM_PORT=5460
       - SPP_PLATFORM_CERTIFICATE_FILE=/skywalking/spp-platform.crt
@@ -65,17 +60,3 @@ services:
       - SW_RECEIVER_GRPC_SSL_ENABLED=true
       - SW_RECEIVER_GRPC_SSL_KEY_PATH=/skywalking/spp-platform.key
       - SW_RECEIVER_GRPC_SSL_CERT_CHAIN_PATH=/skywalking/spp-platform.crt
-  es-server:
-    image: blacktop/elasticsearch:7.0.0
-    container_name: es-server
-    hostname: es-server
-    cap_add:
-      - IPC_LOCK
-    ulimits:
-      memlock:
-        soft: -1
-        hard: -1
-      nofile:
-        soft: 65536
-        hard: 65536
-    command: -E cluster.name=es-skywalking -E path.data=data -E path.logs=logs -E network.host=0.0.0.0 -E node.name=node-1
diff --git a/e2e/spp-probe.yml b/e2e/spp-probe.yml
@@ -12,4 +12,8 @@ skywalking:
     class_cache_mode: "FILE"
     service_name: "spp"
   collector:
-    backend_service: "skywalking-oap:11800"
+    backend_service: "skywalking-oap:11800"
+  plugin:
+    toolkit:
+      log:
+        transmit_formatted: false
diff --git a/sourceplusplus/SourcePlusPlus.py b/sourceplusplus/SourcePlusPlus.py
@@ -133,3 +133,9 @@ def __register_remotes(self, eb, reply_address, status):
                 LiveInstrumentCommand.from_json(json.dumps(msg["body"])), LiveInstrumentType.LOG
             )
         )
+        eb.register_handler(
+            address="spp.probe.command.live-meter-remote:" + self.probe_config["spp"]["probe_id"],
+            handler=lambda msg: self.instrument_remote.handle_instrument_command(
+                LiveInstrumentCommand.from_json(json.dumps(msg["body"])), LiveInstrumentType.METER
+            )
+        )
diff --git a/sourceplusplus/control/ContextReceiver.py b/sourceplusplus/control/ContextReceiver.py
@@ -2,7 +2,6 @@
 import threading
 import time
 import traceback
-
 from skywalking import config, agent
 from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
 from skywalking.protocol.logging.Logging_pb2 import LogData, LogDataBody, TextLog, TraceContext, LogTags
@@ -12,6 +11,7 @@
 from sourceplusplus.control.LiveInstrumentRemote import LiveInstrumentRemote
 from sourceplusplus.models.instrument.LiveBreakpoint import LiveBreakpoint
 from sourceplusplus.models.instrument.LiveLog import LiveLog
+from sourceplusplus.models.instrument.LiveMeter import LiveMeter
 
 
 def try_find(var, globals, locals):
@@ -21,7 +21,16 @@ def try_find(var, globals, locals):
         return globals[var]
 
 
-def do_log(live_log_id, globals, locals):
+def apply_meter(live_meter_id, globals, locals):
+    live_meter: LiveMeter = LiveInstrumentRemote.instruments[live_meter_id][1]
+    if live_meter.throttle.is_rate_limited():
+        return
+    if live_meter.condition is not None and not eval(live_meter.condition, globals, locals):
+        return
+    pass
+
+
+def apply_log(live_log_id, globals, locals):
     live_log: LiveLog = LiveInstrumentRemote.instruments[live_log_id][1]
     if live_log.throttle.is_rate_limited():
         return
@@ -80,7 +89,7 @@ def do_log(live_log_id, globals, locals):
         })
 
 
-def do_breakpoint(live_breakpoint_id, globals, locals):
+def apply_breakpoint(live_breakpoint_id, globals, locals):
     live_breakpoint: LiveBreakpoint = LiveInstrumentRemote.instruments[live_breakpoint_id][1]
     if live_breakpoint.throttle.is_rate_limited():
         return
@@ -94,7 +103,7 @@ def do_breakpoint(live_breakpoint_id, globals, locals):
         for key in locals:
             var = try_find(key, globals, locals)
             tag = StringTag(json.dumps({
-                key: str(var),  # don't str everything
+                key: str(var),  # todo: don't str everything
                 "@class": str(type(var)),
                 "@identity": id(var)
             }))
diff --git a/sourceplusplus/control/LiveInstrumentRemote.py b/sourceplusplus/control/LiveInstrumentRemote.py
@@ -1,15 +1,15 @@
 import json
 import sys
 import threading
-
+from nopdb import nopdb
 from vertx import EventBus
 
-from nopdb import nopdb
 from sourceplusplus.models.command.CommandType import CommandType
 from sourceplusplus.models.command.LiveInstrumentCommand import LiveInstrumentCommand
 from sourceplusplus.models.command.LiveInstrumentContext import LiveInstrumentContext
 from sourceplusplus.models.instrument.LiveBreakpoint import LiveBreakpoint
 from sourceplusplus.models.instrument.LiveLog import LiveLog
+from sourceplusplus.models.instrument.LiveMeter import LiveMeter
 from sourceplusplus.models.instrument.common.LiveInstrumentType import LiveInstrumentType
 from sourceplusplus.models.instrument.common.LiveSourceLocation import LiveSourceLocation
 
@@ -25,55 +25,50 @@ def __init__(self, eb: EventBus):
         LiveInstrumentRemote.dbg.start()
         threading.settrace(sys.gettrace())
 
-    def add_live_log(self, context: LiveInstrumentContext):
-        for i in context.instruments:
-            live_log = LiveLog.from_json(i)
-            bp = LiveInstrumentRemote.dbg.breakpoint(file=live_log.location.source, line=live_log.location.line)
-            LiveInstrumentRemote.instruments[live_log.id] = [bp, live_log]
-            bp.exec("import sourceplusplus.control.ContextReceiver as ContextReceiver\n"
-                    "ContextReceiver.do_log('" + live_log.id + "',globals(),locals())")
-            self.eb.publish(address="spp.platform.status.live-log-applied", body=json.loads(i))
-
-    def remove_live_log(self, context: LiveInstrumentContext):
-        print("Removing live log(s)")
-        for i in context.instruments:
-            live_log = LiveLog.from_json(i)
-            try:
-                LiveInstrumentRemote.dbg.remove_callback(LiveInstrumentRemote.instruments.pop(live_log.id)[0]._handle)
-            except KeyError:
-                pass
-        for i in context.locations:
-            loc = LiveSourceLocation.from_json(i)
-            delete = []
-            for key, val in LiveInstrumentRemote.instruments.items():
-                if isinstance(val[1], LiveLog) and val[1].location == loc:
-                    delete.append(key)
-            for i in delete:
-                del LiveInstrumentRemote.instruments[i]
-
-    def add_live_breakpoint(self, context: LiveInstrumentContext):
-        print("Adding live breakpoint(s)")
+    def add_live_instrument(self, context: LiveInstrumentContext, instrument_type: LiveInstrumentType):
         for i in context.instruments:
-            live_bp = LiveBreakpoint.from_json(i)
-            bp = LiveInstrumentRemote.dbg.breakpoint(file=live_bp.location.source, line=live_bp.location.line)
-            LiveInstrumentRemote.instruments[live_bp.id] = [bp, live_bp]
-            bp.exec("import sourceplusplus.control.ContextReceiver as ContextReceiver\n"
-                    "ContextReceiver.do_breakpoint('" + live_bp.id + "',globals(),locals())")
-            self.eb.publish(address="spp.platform.status.live-breakpoint-applied", body=json.loads(i))
+            if instrument_type == LiveInstrumentType.BREAKPOINT:
+                live_instrument = LiveBreakpoint.from_json(i)
+            elif instrument_type == LiveInstrumentType.LOG:
+                live_instrument = LiveLog.from_json(i)
+            else:
+                live_instrument = LiveMeter.from_json(i)
+            bp = LiveInstrumentRemote.dbg.breakpoint(
+                file=live_instrument.location.source,
+                line=live_instrument.location.line
+            )
+            LiveInstrumentRemote.instruments[live_instrument.id] = [bp, live_instrument]
+            if instrument_type == LiveInstrumentType.BREAKPOINT:
+                bp.exec("import sourceplusplus.control.ContextReceiver as ContextReceiver\n"
+                        "ContextReceiver.apply_breakpoint('" + live_instrument.id + "',globals(),locals())")
+                self.eb.publish(address="spp.platform.status.live-breakpoint-applied", body=json.loads(i))
+            elif instrument_type == LiveInstrumentType.LOG:
+                bp.exec("import sourceplusplus.control.ContextReceiver as ContextReceiver\n"
+                        "ContextReceiver.apply_log('" + live_instrument.id + "',globals(),locals())")
+                self.eb.publish(address="spp.platform.status.live-log-applied", body=json.loads(i))
+            else:
+                bp.exec("import sourceplusplus.control.ContextReceiver as ContextReceiver\n"
+                        "ContextReceiver.apply_meter('" + live_instrument.id + "',globals(),locals())")
+                self.eb.publish(address="spp.platform.status.live-meter-applied", body=json.loads(i))
 
-    def remove_live_breakpoint(self, context: LiveInstrumentContext):
-        print("Removing live breakpoint(s)")
+    def remove_live_instrument(self, context: LiveInstrumentContext, type: LiveInstrumentType):
+        print("Removing live instrument(s)")
         for i in context.instruments:
-            live_bp = LiveBreakpoint.from_json(i)
+            if type == LiveInstrumentType.BREAKPOINT:
+                instrument = LiveBreakpoint.from_json(i)
+            elif type == LiveInstrumentType.LOG:
+                instrument = LiveLog.from_json(i)
+            else:
+                instrument = LiveMeter.from_json(i)
             try:
-                LiveInstrumentRemote.dbg.remove_callback(LiveInstrumentRemote.instruments.pop(live_bp.id)[0]._handle)
+                LiveInstrumentRemote.dbg.remove_callback(LiveInstrumentRemote.instruments.pop(instrument.id)[0]._handle)
             except KeyError:
                 pass
         for i in context.locations:
             loc = LiveSourceLocation.from_json(i)
             delete = []
             for key, val in LiveInstrumentRemote.instruments.items():
-                if isinstance(val[1], LiveBreakpoint) and val[1].location == loc:
+                if val[1].type == type and val[1].location == loc:
                     delete.append(key)
             for i in delete:
                 del LiveInstrumentRemote.instruments[i]
@@ -82,11 +77,10 @@ def handle_instrument_command(self, command: LiveInstrumentCommand, instrument_t
         print("Received command: " + command.command_type)
         if command.command_type == CommandType.ADD_LIVE_INSTRUMENT:
             if instrument_type == LiveInstrumentType.BREAKPOINT:
-                self.add_live_breakpoint(command.context)
+                self.add_live_instrument(command.context, LiveInstrumentType.BREAKPOINT)
+            elif instrument_type == LiveInstrumentType.LOG:
+                self.add_live_instrument(command.context, LiveInstrumentType.LOG)
             else:
-                self.add_live_log(command.context)
+                self.add_live_instrument(command.context, LiveInstrumentType.METER)
         elif command.command_type == CommandType.REMOVE_LIVE_INSTRUMENT:
-            if instrument_type == LiveInstrumentType.BREAKPOINT:
-                self.remove_live_breakpoint(command.context)
-            else:
-                self.remove_live_log(command.context)
+            self.remove_live_instrument(command.context, instrument_type)
diff --git a/sourceplusplus/models/instrument/LiveMeter.py b/sourceplusplus/models/instrument/LiveMeter.py
@@ -0,0 +1,32 @@
+import json
+
+import humps
+
+from .common.LiveInstrument import LiveInstrument
+from .common.LiveInstrumentType import LiveInstrumentType
+from .common.LiveSourceLocation import LiveSourceLocation
+from .common.throttle.HitThrottle import HitThrottle
+from .common.throttle.ThrottleStep import ThrottleStep
+
+
+class LiveMeter(LiveInstrument):
+    def __init__(self, location: LiveSourceLocation):
+        super().__init__(location)
+        self.hit_limit = -1
+        self.type = LiveInstrumentType.METER
+
+    @classmethod
+    def from_json(cls, json_str):
+        json_dict = humps.decamelize(json.loads(json_str))
+        # todo: easier way to convert
+        location = LiveSourceLocation(json_dict["location"]["source"], json_dict["location"]["line"])
+        meter = LiveMeter(location)
+        for key in json_dict:
+            setattr(meter, key, json_dict[key])
+        meter.location = LiveSourceLocation(json_dict["location"]["source"], json_dict["location"]["line"])
+
+        if "throttle" in json_dict:
+            meter.throttle = HitThrottle(json_dict["throttle"]["limit"], ThrottleStep(json_dict["throttle"]["step"]))
+
+        meter.type = LiveInstrumentType.METER
+        return meter
diff --git a/sourceplusplus/models/instrument/common/LiveInstrumentType.py b/sourceplusplus/models/instrument/common/LiveInstrumentType.py
@@ -4,3 +4,4 @@
 class LiveInstrumentType(str, Enum):
     BREAKPOINT = "BREAKPOINT"
     LOG = "LOG"
+    METER = "METER"

Original file line number	Diff line number	Diff line change
`@@ -133,3 +133,9 @@ def __register_remotes(self, eb, reply_address, status):`
`133`	`133`	`LiveInstrumentCommand.from_json(json.dumps(msg["body"])), LiveInstrumentType.LOG`
`134`	`134`	`)`
`135`	`135`	`)`
	`136`	`+ eb.register_handler(`
	`137`	`+ address="spp.probe.command.live-meter-remote:" + self.probe_config["spp"]["probe_id"],`
	`138`	`+ handler=lambda msg: self.instrument_remote.handle_instrument_command(`
	`139`	`+ LiveInstrumentCommand.from_json(json.dumps(msg["body"])), LiveInstrumentType.METER`
	`140`	`+ )`
	`141`	`+ )`