ActivityPub/workloadGenerator.go

package main

import (
	"bytes"
	"crypto/tls"
	"encoding/base64"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"net/http"
	"os"
	"strconv"
	"time"

	"git.nutfactory.org/hoernschen/ActivityPub/config"
	"git.nutfactory.org/hoernschen/ActivityPub/entities/activity"
	"git.nutfactory.org/hoernschen/ActivityPub/entities/object"
	"git.nutfactory.org/hoernschen/ActivityPub/entities/user"
	"git.nutfactory.org/hoernschen/ActivityPub/utils"
)

var BaseLineTest = false

var systemParamsIndex = 0

type SystemParams struct {
	Id                    string
	BytesToSend           int
	MessagesPerSecond     float32
	Distribution          map[string][]string
	Packetloss            int
	MinutesNotAvailable   int
	Consensus             bool
	AuthentificationCheck bool
	Signing               bool
	Encryption            bool
}

var httpString string
var users = []map[string][]string{
	map[string][]string{
		"143.93.38.208": []string{
			"user1",
			"user2",
			"user3",
			"user4",
			"user5",
			"user6",
		},
	},
	map[string][]string{
		"143.93.38.207": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.208": []string{
			"user1",
		},
		"143.93.38.209": []string{
			"user1",
		},
	},
	map[string][]string{
		"143.93.38.207": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.208": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.209": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
	},
	map[string][]string{
		"localhost": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
	},
}

/*
var servers = []string{
	"143.93.38.207",
	"143.93.38.208",
	"143.93.38.209",
}
*/
var servers = []string{
	"localhost",
}

var systemParams = []SystemParams{
	SystemParams{
		Id:                    "111110101",
		BytesToSend:           280,
		MessagesPerSecond:     1.0,
		Distribution:          users[3],
		Packetloss:            1,
		MinutesNotAvailable:   0,
		Consensus:             true,
		AuthentificationCheck: true,
		Signing:               true,
		Encryption:            true,
	},
	/*
		SystemParams{
			Id:                    "111110101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "011110101",
			BytesToSend:           8,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "211110101",
			BytesToSend:           512,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "101110101",
			BytesToSend:           280,
			MessagesPerSecond:     0.1,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "121110101",
			BytesToSend:           280,
			MessagesPerSecond:     10.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "110110101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[0],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "112110101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[2],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111010101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            0,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111210101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            20,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		/*
			SystemParams{
				Id:                    "111120000",
				BytesToSend:           280,
				MessagesPerSecond:     1.0,
				Distribution:          users[1],
				Packetloss:            1,
				MinutesNotAvailable:   1,
				Consensus:             true,
				AuthentificationCheck: true,
				Signing:               true,
				Encryption:            true,
			},
	*/
}

var userIds []string
var accessTokens map[string]string
var collectionId string

func main() {
	http.DefaultTransport.(*http.Transport).TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	httpString = "https"
	iteration := 0
	if BaseLineTest {
		file, err := os.Create(fmt.Sprintf("%s Baseline Measurement ActivityPub.csv", strconv.FormatInt(time.Now().Unix(), 10)))
		if err != nil {
			log.Fatalf("Error Creating CSV: %s", err)
		}
		defer file.Close()
		writer := csv.NewWriter(file)
		defer writer.Flush()
		err = writer.Write([]string{"Iteration", "Start", "End"})
		if err != nil {
			log.Fatalf("Error in Writing CSV: %s", err)
		}
		for iteration < 10 {
			start := time.Now().Unix()
			time.Sleep(2 * time.Minute)
			end := time.Now().Unix()
			iteration++
			err = writer.Write([]string{strconv.Itoa(iteration), strconv.FormatInt(start, 10), strconv.FormatInt(end, 10)})
			if err != nil {
				log.Fatalf("Error in Writing CSV: %s", err)
			}
		}
	} else {
		for _, systemParam := range systemParams {
			file, err := os.Create(fmt.Sprintf("%s %s Measurement ActivityPub.csv", strconv.FormatInt(time.Now().Unix(), 10), systemParam.Id))
			if err != nil {
				log.Fatalf("Error Creating CSV: %s", err)
			}
			writer := csv.NewWriter(file)
			err = writer.Write([]string{"Iteration", "Start", "End", "Actions Send"})
			if err != nil {
				log.Fatalf("Error in Writing CSV: %s", err)
			}
			millisecondsToWait := 1000 / systemParam.MessagesPerSecond
			iteration = 0
			for iteration < 30 {
				err := setUp(systemParam)
				if err != nil {
					log.Fatalf("Error in SetUp: %s", err)
				}
				b := make([]byte, systemParam.BytesToSend)
				_, err = rand.Read(b)
				message := base64.RawStdEncoding.EncodeToString(b)
				start := time.Now()
				end := start.Add(2 * time.Minute).Unix()
				log.Printf("Id: %s - Iteration: %s - Start: %s - End: %s", systemParam.Id, strconv.Itoa(iteration), strconv.FormatInt(start.Unix(), 10), strconv.FormatInt(end, 10))
				actionsSend := 0
				for time.Now().Unix() < end {
					time.Sleep(time.Duration(millisecondsToWait) * time.Millisecond)

					err = sendMessage(message)
					if err != nil {
						log.Fatalf("Error sending Message %s", err)
					}

					actionsSend++
				}
				iteration++
				err = writer.Write([]string{strconv.Itoa(iteration), strconv.FormatInt(start.Unix(), 10), strconv.FormatInt(end, 10), strconv.Itoa(actionsSend)})
				if err != nil {
					log.Fatalf("Error in Writing CSV: %s", err)
				}
				time.Sleep(1 * time.Second)
				err = reset()
				if err != nil {
					log.Fatalf("Error in Reset: %s", err)
				}
				time.Sleep(1 * time.Second)
			}
			writer.Flush()
			file.Close()
		}
	}
}

func sendMessage(message string) (err error) {
	userId := userIds[0]
	accessToken := accessTokens[userId]
	requestUrl := fmt.Sprintf("%soutbox/", userId)
	messageObject := object.Object{
		Type:    "Note",
		Content: message,
	}
	reqBody, err := json.Marshal(messageObject)
	if err != nil {
		return
	}
	client := &http.Client{}
	var req *http.Request
	req, err = http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{utils.GetContentTypeString()}
	req.Header["Authorization"] = []string{fmt.Sprintf("Bearer %s", accessToken)}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		handleError(res)
	}
	return
}

func setParams(systemParamsToUse SystemParams) (err error) {
	serverNotAvailableIndex := 1
	for i, server := range servers {
		minutesNotAvailable := 0
		if serverNotAvailableIndex == i {
			minutesNotAvailable = systemParamsToUse.MinutesNotAvailable
		}
		requestUrl := fmt.Sprintf("%s://%s/setparams", httpString, server)
		request := config.SetParamBody{
			Packetloss:            systemParamsToUse.Packetloss,
			UnavailableTill:       time.Now().Add(time.Duration(minutesNotAvailable) * time.Minute).Unix(),
			Consensus:             systemParamsToUse.Consensus,
			AuthentificationCheck: systemParamsToUse.AuthentificationCheck,
			Signing:               systemParamsToUse.Signing,
			Encryption:            systemParamsToUse.Encryption,
		}
		var reqBody []byte
		reqBody, err = json.Marshal(request)
		if err != nil {
			return
		}
		client := &http.Client{Timeout: 2 * time.Second}
		var req *http.Request
		req, err = http.NewRequest(http.MethodGet, requestUrl, bytes.NewBuffer(reqBody))
		if err != nil {
			return
		}
		req.Header["Content-Type"] = []string{"application/json"}
		var res *http.Response
		res, err = client.Do(req)
		if err != nil {
			return
		}
		if res.StatusCode != http.StatusOK {
			handleError(res)
		}
	}
	return
}

func setUp(systemParamsToUse SystemParams) (err error) {
	accessTokens = make(map[string]string)
	err = createUsers(systemParamsToUse.Distribution)
	if err != nil {
		log.Printf("Error in User-Creation: %s", err)
		return
	}

	// Follow User
	log.Printf("Userids: %s", userIds)
	err = followUser(userIds[0], userIds[1:])
	if err != nil {
		log.Printf("Error follow User: %s", err)
		return
	}

	err = setParams(systemParamsToUse)
	return
}

func reset() (err error) {
	userIds = []string{}
	accessTokens = make(map[string]string)
	for _, server := range servers {
		requestUrl := fmt.Sprintf("%s://%s/reset", httpString, server)
		client := &http.Client{Timeout: 2 * time.Second}
		var req *http.Request
		req, err = http.NewRequest(http.MethodGet, requestUrl, bytes.NewBuffer(nil))
		if err != nil {
			return
		}
		req.Header["Content-Type"] = []string{"application/json"}
		var res *http.Response
		res, err = client.Do(req)
		if err != nil {
			return
		}
		if res.StatusCode != http.StatusOK {
			handleError(res)
		}
	}
	return
}

func followUser(userIdToFollow string, userIdsThatFollow []string) (err error) {
	for _, userId := range userIdsThatFollow {
		log.Printf("UserIdThatFollow: %s", userId)
		accessToken := accessTokens[userId]
		requestUrl := fmt.Sprintf("%soutbox/", userId)
		followActivity := activity.Activity{
			Type:  "Follow",
			Actor: userId,
			To:    userIdToFollow,
		}
		var reqBody []byte
		reqBody, err = json.Marshal(followActivity)
		if err != nil {
			return
		}
		client := &http.Client{Timeout: 2 * time.Second}
		var req *http.Request
		req, err = http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
		if err != nil {
			return
		}
		req.Header["Content-Type"] = []string{utils.GetContentTypeString()}
		req.Header["Authorization"] = []string{fmt.Sprintf("Bearer %s", accessToken)}
		var res *http.Response
		res, err = client.Do(req)
		if err != nil {
			return
		}
		if res.StatusCode != http.StatusOK {
			err = utils.HandleHTTPError(res)
			return
		}
		//time.Sleep(time.Duration(1) * time.Second)
	}
	return
}

func createUsers(serverUserMap map[string][]string) (err error) {
	log.Println("Create Users")
	for server, usersToCreate := range serverUserMap {
		for _, userToCreate := range usersToCreate {
			var userId string
			var accessToken string
			userId, accessToken, err = createUser(userToCreate, server)
			if err != nil {
				return
			}
			if userId != "" && accessToken != "" {
				log.Printf("%s created - AccessToken: %s", userId, accessToken)
				accessTokens[userId] = accessToken
				userIds = append(userIds, userId)
			}
			time.Sleep(time.Duration(20) * time.Millisecond)
		}
	}
	return
}

func createUser(userToCreate string, homeserver string) (userId string, accessToken string, err error) {
	requestUrl := fmt.Sprintf("%s://%s/register", httpString, homeserver)
	request := user.RegisterRequest{
		Username: userToCreate,
		Password: "password",
	}
	reqBody, err := json.Marshal(request)
	if err != nil {
		return
	}
	client := &http.Client{Timeout: 2 * time.Second}
	req, err := http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{utils.GetContentTypeString()}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		err = utils.HandleHTTPError(res)
		return
	} else {
		response := user.RegisterResponse{}
		decoder := json.NewDecoder(res.Body)
		err = decoder.Decode(&response)
		if err != nil {
			return
		}
		userId = response.Actor
		accessToken = response.Token
	}
	return
}

func handleError(res *http.Response) {
	err := utils.HandleHTTPError(res)
	if err != nil {
		log.Printf("%s", err)
	}
}