Export metric port to 8008 in Dockerfile

.gitignore

@@ -1,2 +1,4 @@
 nimbledeps/*
 *.exe
+stew/*
+nim-stew-master/*

Dockerfile

@@ -0,0 +1,30 @@
+# Create the build image
+FROM nimlang/nim:1.6.18 as build
+
+# Copy the wls files to the production image
+WORKDIR /node
+COPY . .
+
+RUN git config --global http.sslVerify false
+
+RUN nimble install -dy
+
+RUN nimble c -d:chronicles_colors=None --threads:on -d:metrics -d:libp2p_network_protocols_metrics  -d:release main
+
+
+FROM nimlang/nim:1.6.18
+
+RUN apt-get install cron -y
+
+WORKDIR /node
+
+COPY --from=build /node/main /node/main
+
+COPY cron_runner.sh .
+
+RUN chmod +x cron_runner.sh
+RUN chmod +x main
+
+EXPOSE 5000 8008
+
+ENTRYPOINT ["./cron_runner.sh"]

cron_runner.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+
+if [ "$#" -ne 2 ]; then
+    echo "Usage: $0 <minutes> <hours>"
+    exit 1
+fi
+
+minutes="$1"
+hours="$2"
+
+cron_expression="$minutes $hours * * *"
+
+cron_job_file="/etc/cron.d/my-cron-job"
+echo -e "$cron_expression /node/main  > /proc/1/fd/1 2>&1 \n" > "$cron_job_file"
+
+echo "Cron job file created at $cron_job_file"
+
+env >> /etc/environment
+
+crontab /etc/cron.d/my-cron-job
+
+cron -f

main.nim

@@ -2,7 +2,7 @@ import stew/endians2, stew/byteutils, tables, strutils, os
 import libp2p, libp2p/protocols/pubsub/rpc/messages
 import libp2p/muxers/mplex/lpchannel, libp2p/protocols/ping
 import chronos
-import sequtils, hashes, math, metrics
+import sequtils, hashes, math, metrics, metrics/chronos_httpserver
 from times import getTime, toUnix, fromUnix, `-`, initTime, `$`, inMilliseconds
 from nativesockets import getHostname
 
@@ -11,20 +11,44 @@ const chunks = 1
 proc msgIdProvider(m: Message): Result[MessageId, ValidationResult] =
   return ok(($m.data.hash).toBytes())
 
+proc startMetricsServer(
+    serverIp: IpAddress, serverPort: Port
+): Result[MetricsHttpServerRef, string] =
+  info "Starting metrics HTTP server", serverIp = $serverIp, serverPort = $serverPort
+
+  let metricsServerRes = MetricsHttpServerRef.new($serverIp, serverPort)
+  if metricsServerRes.isErr():
+    return err("metrics HTTP server start failed: " & $metricsServerRes.error)
+
+  let server = metricsServerRes.value
+  try:
+    waitFor server.start()
+  except CatchableError:
+    return err("metrics HTTP server start failed: " & getCurrentExceptionMsg())
+
+  info "Metrics HTTP server started", serverIp = $serverIp, serverPort = $serverPort
+  ok(metricsServerRes.value)
+
 proc main {.async.} =
   let
     hostname = getHostname()
-    myId = parseInt(hostname[4..^1])
+    myId = parseInt(getEnv("PEERNUMBER"))
+    msg_rate = parseInt(getEnv("MSGRATE"))
+    msg_size = parseInt(getEnv("MSGSIZE"))
+
     #publisherCount = client.param(int, "publisher_count")
-    publisherCount = 10
+    publisherCount = parseInt(getEnv("PEERS"))
     isPublisher = myId <= publisherCount
     #isAttacker = (not isPublisher) and myId - publisherCount <= client.param(int, "attacker_count")
     isAttacker = false
     rng = libp2p.newRng()
     #randCountry = rng.rand(distribCumSummed[^1])
     #country = distribCumSummed.find(distribCumSummed.filterIt(it >= randCountry)[0])
+  echo "Hostname: ", hostname
   let
-    address = initTAddress("0.0.0.0:5000")
+    myport = 5000 + parseInt(getEnv("PEERNUMBER"))
+    myaddress = "0.0.0.0:" & $myport
+    address = initTAddress(myaddress)
     switch =
       SwitchBuilder
         .new()
@@ -32,7 +56,7 @@ proc main {.async.} =
         .withRng(rng)
         #.withYamux()
         .withMplex()
-        .withMaxConnections(10000)
+        .withMaxConnections(250)
         .withTcpTransport(flags = {ServerFlags.TcpNoDelay})
         #.withPlainText()
         .withNoise()
@@ -45,16 +69,20 @@ proc main {.async.} =
       anonymize = true,
       )
     pingProtocol = Ping.new(rng=rng)
-  gossipSub.parameters.floodPublish = false
+  # Metrics
+  echo "Starting metrics HTTP server"
+  let metricsServer = startMetricsServer(parseIpAddress("0.0.0.0"), Port(8008))
+
+  gossipSub.parameters.floodPublish = true
   #gossipSub.parameters.lazyPushThreshold = 1_000_000_000
   #gossipSub.parameters.lazyPushThreshold = 0
   gossipSub.parameters.opportunisticGraftThreshold = -10000
-  gossipSub.parameters.heartbeatInterval = 700.milliseconds
-  gossipSub.parameters.pruneBackoff = 3.seconds
-  gossipSub.parameters.gossipFactor = 0.05
-  gossipSub.parameters.d = 8
-  gossipSub.parameters.dLow = 6
-  gossipSub.parameters.dHigh = 12
+  gossipSub.parameters.heartbeatInterval = 1.seconds
+  gossipSub.parameters.pruneBackoff = 60.seconds
+  gossipSub.parameters.gossipFactor = 0.25
+  gossipSub.parameters.d = 6
+  gossipSub.parameters.dLow = 4
+  gossipSub.parameters.dHigh = 8
   gossipSub.parameters.dScore = 6
   gossipSub.parameters.dOut = 6 div 2
   gossipSub.parameters.dLazy = 6
@@ -96,14 +124,12 @@ proc main {.async.} =
   switch.mount(gossipSub)
   switch.mount(pingProtocol)
   await switch.start()
-  #TODO
   #defer: await switch.stop()
 
   echo "Listening on ", switch.peerInfo.addrs
   echo myId, ", ", isPublisher, ", ", switch.peerInfo.peerId
-
-  var peersInfo = toSeq(1..parseInt(getEnv("PEERS")))
-  rng.shuffle(peersInfo)
+  echo "Waiting 60 seconds for node building..."
+  await sleepAsync(60.seconds)
 
   proc pinger(peerId: PeerId) {.async.} =
     try:
@@ -120,41 +146,58 @@ proc main {.async.} =
 
   let connectTo = parseInt(getEnv("CONNECTTO"))
   var connected = 0
-  for peerInfo in peersInfo:
-    if connected >= connectTo: break
-    let tAddress = "peer" & $peerInfo & ":5000"
-    echo tAddress
-    let addrs = resolveTAddress(tAddress).mapIt(MultiAddress.init(it).tryGet())
+  let tAddress = "nimp2p-service:5000"
+  var addrs: seq[MultiAddress]
+
+  echo "Trying to resolve ", tAddress
+  while true:
     try:
-      let peerId = await switch.connect(addrs[0], allowUnknownPeerId=true).wait(5.seconds)
-      #asyncSpawn pinger(peerId)
-      connected.inc()
+      addrs = resolveTAddress(tAddress).mapIt(MultiAddress.init(it).tryGet())
+      echo tAddress, " resolved: ", addrs
+      break  # Break out of the loop on successful resolution
     except CatchableError as exc:
-      echo "Failed to dial", exc.msg
+      echo "Failed to resolve address:", exc.msg
+      echo "Waiting 15 seconds..."
+      await sleepAsync(15.seconds)
+
+  rng.shuffle(addrs)
+  var index = 0
+  while true:
+    if connected >= connectTo: break
+    while true:
+      try:
+        echo "Trying to connect to ", addrs[index]
+        let peerId = await switch.connect(addrs[index], allowUnknownPeerId=true).wait(5.seconds)
+        #asyncSpawn pinger(peerId)
+        connected.inc()
+        index.inc()
+        echo "Connected!"
+        break
+      except CatchableError as exc:
+        echo "Failed to dial", exc.msg
+        echo "Waiting 15 seconds..."
+        await sleepAsync(15.seconds)
 
   #let
   #  maxMessageDelay = client.param(int, "max_message_delay")
   #  warmupMessages = client.param(int, "warmup_messages")
   #startOfTest = Moment.now() + milliseconds(warmupMessages * maxMessageDelay div 2)
 
-  await sleepAsync(10.seconds)
   echo "Mesh size: ", gossipSub.mesh.getOrDefault("test").len
 
-  for msg in 0 ..< 10:#client.param(int, "message_count"):
-    await sleepAsync(12.seconds)
-    if msg mod publisherCount == myId - 1:
-    #if myId == 1:
+  let turnToPublish = parseInt(getHostname()[4..^1])
+  echo "Publishing turn is: ", turnToPublish
+  for msg in 0 ..< 10000:#client.param(int, "message_count"):
+    await sleepAsync(msg_rate)
+    if msg mod publisherCount == turnToPublish:
+      echo "Sending message at: " ,times.getTime()
       let
         now = getTime()
         nowInt = seconds(now.toUnix()) + nanoseconds(times.nanosecond(now))
-      #var nowBytes = @(toBytesLE(uint64(nowInt.nanoseconds))) & newSeq[byte](500_000 div chunks)
-      var nowBytes = @(toBytesLE(uint64(nowInt.nanoseconds))) & newSeq[byte](50)
-      #echo "sending ", uint64(nowInt.nanoseconds)
-      for chunk in 0..<chunks:
-        nowBytes[10] = byte(chunk)
-        doAssert((await gossipSub.publish("test", nowBytes)) > 0)
+      var nowBytes = @(toBytesLE(uint64(nowInt.nanoseconds))) & newSeq[byte](msg_size)
+      doAssert((await gossipSub.publish("test", nowBytes)) > 0)
 
   #echo "BW: ", libp2p_protocols_bytes.value(labelValues=["/meshsub/1.1.0", "in"]) + libp2p_protocols_bytes.value(labelValues=["/meshsub/1.1.0", "out"])
   #echo "DUPS: ", libp2p_gossipsub_duplicate.value(), " / ", libp2p_gossipsub_received.value()
 
-waitFor(main())
+waitFor(main())

runner.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+custom_network_name="my_custom_network"
+num_peers=10
+
+if ! docker network inspect "$custom_network_name" >/dev/null 2>&1; then
+  docker network create "$custom_network_name"
+  docker network create --attachable --driver bridge "$custom_network_name"
+fi
+
+for ((i = 0; i < num_peers; i++)); do
+    # Construct the hostname (e.g., peer1, peer2, ...)
+    hostname="pod-$i"
+
+    # Run the Docker container with the current hostname
+    docker run -e PEERSPERPOD="1" -e PEERS="10" -e CONNECTTO="5" -e MSGRATE="1000" -e MSGSIZE="1000" -e PEERNUMBER="0" --hostname="$hostname" --network="$custom_network_name" dst-test-node &
+
+done

shadow/run.sh

@@ -1,8 +0,0 @@
-#!/bin/sh
-
-set -e
-
-nim c -d:chronicles_colors=None --threads:on -d:metrics -d:libp2p_network_protocols_metrics -d:release main
-rm -rf shadow.data/
-shadow shadow.yaml
-grep -rne 'milliseconds\|BW' shadow.data/ > latencies

test_node.nimble

@@ -8,5 +8,5 @@ license       = "MIT"
 skipDirs      = @[]
 
 requires "nim >= 1.6.0",
-         "libp2p",
-         "ggplotnim"
+          "libp2p",
+          "ggplotnim"