Upgrade Sonyflake to v2 (#66)

* Add v2

* Introduce staticcheck

* Introduce golangci-lint

* Add error checks

* Add error checks

* Lint v2 code

* Improve CI trigger

* Use io.ReadAll

* Use int64

* Remove NewSonyflake

* Fix errors

* v2: Change MachineID, sequence, and AmazonEC2MachineID to int; update all usage and tests for type consistency

* docs: update Settings struct in README to use int for MachineID and CheckMachineID (v2)

* docs(v2): clarify Settings, StartTime, MachineID, and CheckMachineID comments and update README links and explanations

* docs(v2/mock): improve comments and docstrings for mock implementations

* docs(types): unify and clarify package and type docstrings for types.go in v1 and v2

* test(v2): refactor and modernize tests, improve error assertions, and update mocks for v2

* test(v2): normalize whitespace in pseudoSleep calls for consistency

* feat(v2): add configurable TimeUnit and refactor time handling for So… (#67)

* feat(v2): add configurable TimeUnit and refactor time handling for Sonyflake v2

* test(v2): add ToTime tests, clarify TimeUnit behavior, and update docs for v2

* gofmt
This commit is contained in:
Yoshiyuki Mineo
2025-05-05 10:53:08 +09:00
committed by GitHub
parent 8d195df6f7
commit 357b2ee5f0
19 changed files with 896 additions and 56 deletions
+64
View File
@@ -0,0 +1,64 @@
// Package awsutil provides utility functions for using Sonyflake on AWS.
package awsutil
import (
"errors"
"io"
"net"
"net/http"
"os/exec"
"regexp"
"strconv"
"time"
)
func amazonEC2PrivateIPv4() (net.IP, error) {
res, err := http.Get("http://169.254.169.254/latest/meta-data/local-ipv4")
if err != nil {
return nil, err
}
defer res.Body.Close()
body, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
ip := net.ParseIP(string(body))
if ip == nil {
return nil, errors.New("invalid ip address")
}
return ip.To4(), nil
}
// AmazonEC2MachineID retrieves the private IP address of the Amazon EC2 instance
// and returns its lower 16 bits.
// It works correctly on Docker as well.
func AmazonEC2MachineID() (int, error) {
ip, err := amazonEC2PrivateIPv4()
if err != nil {
return 0, err
}
return int(ip[2])<<8 + int(ip[3]), nil
}
// TimeDifference returns the time difference between the localhost and the given NTP server.
func TimeDifference(server string) (time.Duration, error) {
output, err := exec.Command("/usr/sbin/ntpdate", "-q", server).CombinedOutput()
if err != nil {
return time.Duration(0), err
}
re, _ := regexp.Compile("offset (.*) sec")
submatched := re.FindSubmatch(output)
if len(submatched) != 2 {
return time.Duration(0), errors.New("invalid ntpdate output")
}
f, err := strconv.ParseFloat(string(submatched[1]), 64)
if err != nil {
return time.Duration(0), err
}
return time.Duration(f*1000) * time.Millisecond, nil
}
+6
View File
@@ -0,0 +1,6 @@
FROM ubuntu:24.04
ADD ./sonyflake_server /
ENTRYPOINT ["/sonyflake_server"]
EXPOSE 8080
+8
View File
@@ -0,0 +1,8 @@
{
"AWSEBDockerrunVersion": "1",
"Ports": [
{
"ContainerPort": "8080"
}
]
}
+21
View File
@@ -0,0 +1,21 @@
Example
=======
This example runs Sonyflake on AWS Elastic Beanstalk.
Setup
-----
1. Build the cross compiler for linux/amd64 if using other platforms.
```
cd $GOROOT/src && GOOS=linux GOARCH=amd64 ./make.bash
```
2. Build sonyflake_server in the example directory.
```
./linux64_build.sh
```
3. Upload the example directory to AWS Elastic Beanstalk.
+2
View File
@@ -0,0 +1,2 @@
#!/bin/sh
GOOS=linux GOARCH=amd64 go build sonyflake_server.go
+51
View File
@@ -0,0 +1,51 @@
package main
import (
"encoding/json"
"net/http"
"github.com/sony/sonyflake/v2"
"github.com/sony/sonyflake/v2/awsutil"
)
var sf *sonyflake.Sonyflake
func init() {
var st sonyflake.Settings
st.MachineID = awsutil.AmazonEC2MachineID
var err error
sf, err = sonyflake.New(st)
if err != nil {
panic(err)
}
}
func handler(w http.ResponseWriter, r *http.Request) {
id, err := sf.NextID()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
body, err := json.Marshal(sonyflake.Decompose(id))
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header()["Content-Type"] = []string{"application/json; charset=utf-8"}
_, err = w.Write(body)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
func main() {
http.HandleFunc("/", handler)
err := http.ListenAndServe(":8080", nil)
if err != nil {
panic(err)
}
}
+3
View File
@@ -0,0 +1,3 @@
module github.com/sony/sonyflake/v2
go 1.22
+36
View File
@@ -0,0 +1,36 @@
// Package mock offers mock implementations of interfaces defined in types.go.
// This allows complete control over input / output for any given method that consumes a given type.
package mock
import (
"errors"
"net"
"github.com/sony/sonyflake/v2/types"
)
// NewSuccessfulInterfaceAddrs returns a single private IP address.
func NewSuccessfulInterfaceAddrs() types.InterfaceAddrs {
ifat := make([]net.Addr, 0, 1)
ifat = append(ifat, &net.IPNet{IP: []byte{192, 168, 0, 1}, Mask: []byte{255, 0, 0, 0}})
return func() ([]net.Addr, error) {
return ifat, nil
}
}
var ErrFailedToGetAddresses = errors.New("failed to get addresses")
// NewFailingInterfaceAddrs returns an error.
func NewFailingInterfaceAddrs() types.InterfaceAddrs {
return func() ([]net.Addr, error) {
return nil, ErrFailedToGetAddresses
}
}
// NewNilInterfaceAddrs returns an empty slice of addresses.
func NewNilInterfaceAddrs() types.InterfaceAddrs {
return func() ([]net.Addr, error) {
return []net.Addr{}, nil
}
}
+232
View File
@@ -0,0 +1,232 @@
// Package sonyflake implements Sonyflake, a distributed unique ID generator inspired by Twitter's Snowflake.
//
// A Sonyflake ID is composed of
//
// 39 bits for time in units of 10 msec
// 8 bits for a sequence number
// 16 bits for a machine id
package sonyflake
import (
"errors"
"net"
"sync"
"time"
"github.com/sony/sonyflake/v2/types"
)
// These constants are the bit lengths of Sonyflake ID parts.
const (
BitLenTime = 39 // bit length of time
BitLenSequence = 8 // bit length of sequence number
BitLenMachine = 63 - BitLenTime - BitLenSequence // bit length of machine id
)
// Settings configures Sonyflake:
//
// TimeUnit is the time unit of Sonyflake.
// If TimeUnit is 0, the default time unit is used, which is 10 msec.
// TimeUnit must be equal to or greater than 1 msec.
//
// StartTime is the time since which the Sonyflake time is defined as the elapsed time.
// If StartTime is 0, the start time of the Sonyflake instance is set to "2025-01-01 00:00:00 +0000 UTC".
// StartTime must be before the current time.
//
// MachineID returns the unique ID of a Sonyflake instance.
// If MachineID returns an error, the instance will not be created.
// If MachineID is nil, the default MachineID is used, which returns the lower 16 bits of the private IP address.
//
// CheckMachineID validates the uniqueness of a machine ID.
// If CheckMachineID returns false, the instance will not be created.
// If CheckMachineID is nil, no validation is done.
type Settings struct {
TimeUnit time.Duration
StartTime time.Time
MachineID func() (int, error)
CheckMachineID func(int) bool
}
// Sonyflake is a distributed unique ID generator.
type Sonyflake struct {
mutex *sync.Mutex
timeUnit int64
startTime int64
elapsedTime int64
sequence int
machine int
}
var (
ErrStartTimeAhead = errors.New("start time is ahead of now")
ErrNoPrivateAddress = errors.New("no private ip address")
ErrOverTimeLimit = errors.New("over the time limit")
ErrInvalidMachineID = errors.New("invalid machine id")
ErrInvalidTimeUnit = errors.New("invalid time unit")
)
const defaultTimeUnit = 1e7 // nsec, i.e. 10 msec
var defaultInterfaceAddrs = net.InterfaceAddrs
// New returns a new Sonyflake configured with the given Settings.
// New returns an error in the following cases:
// - Settings.StartTime is ahead of the current time.
// - Settings.MachineID returns an error.
// - Settings.CheckMachineID returns false.
func New(st Settings) (*Sonyflake, error) {
if st.StartTime.After(time.Now()) {
return nil, ErrStartTimeAhead
}
sf := new(Sonyflake)
sf.mutex = new(sync.Mutex)
sf.sequence = 1<<BitLenSequence - 1
if st.TimeUnit == 0 {
sf.timeUnit = defaultTimeUnit
} else if st.TimeUnit >= time.Millisecond {
sf.timeUnit = int64(st.TimeUnit)
} else {
return nil, ErrInvalidTimeUnit
}
if st.StartTime.IsZero() {
sf.startTime = sf.toInternalTime(time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC))
} else {
sf.startTime = sf.toInternalTime(st.StartTime)
}
var err error
if st.MachineID == nil {
sf.machine, err = lower16BitPrivateIP(defaultInterfaceAddrs)
} else {
sf.machine, err = st.MachineID()
}
if err != nil {
return nil, err
}
if st.CheckMachineID != nil && !st.CheckMachineID(sf.machine) {
return nil, ErrInvalidMachineID
}
return sf, nil
}
// NextID generates a next unique ID as int64.
// After the Sonyflake time overflows, NextID returns an error.
func (sf *Sonyflake) NextID() (int64, error) {
const maskSequence = 1<<BitLenSequence - 1
sf.mutex.Lock()
defer sf.mutex.Unlock()
current := sf.currentElapsedTime()
if sf.elapsedTime < current {
sf.elapsedTime = current
sf.sequence = 0
} else {
sf.sequence = (sf.sequence + 1) & maskSequence
if sf.sequence == 0 {
sf.elapsedTime++
overtime := sf.elapsedTime - current
sf.sleep(overtime)
}
}
return sf.toID()
}
func (sf *Sonyflake) toInternalTime(t time.Time) int64 {
return t.UTC().UnixNano() / sf.timeUnit
}
func (sf *Sonyflake) currentElapsedTime() int64 {
return sf.toInternalTime(time.Now()) - sf.startTime
}
func (sf *Sonyflake) sleep(overtime int64) {
sleepTime := time.Duration(overtime*sf.timeUnit) -
time.Duration(time.Now().UTC().UnixNano()%sf.timeUnit)
time.Sleep(sleepTime)
}
func (sf *Sonyflake) toID() (int64, error) {
if sf.elapsedTime >= 1<<BitLenTime {
return 0, ErrOverTimeLimit
}
return sf.elapsedTime<<(BitLenSequence+BitLenMachine) |
int64(sf.sequence)<<BitLenMachine |
int64(sf.machine), nil
}
func privateIPv4(interfaceAddrs types.InterfaceAddrs) (net.IP, error) {
as, err := interfaceAddrs()
if err != nil {
return nil, err
}
for _, a := range as {
ipnet, ok := a.(*net.IPNet)
if !ok || ipnet.IP.IsLoopback() {
continue
}
ip := ipnet.IP.To4()
if isPrivateIPv4(ip) {
return ip, nil
}
}
return nil, ErrNoPrivateAddress
}
func isPrivateIPv4(ip net.IP) bool {
// Allow private IP addresses (RFC1918) and link-local addresses (RFC3927)
return ip != nil &&
(ip[0] == 10 || ip[0] == 172 && (ip[1] >= 16 && ip[1] < 32) || ip[0] == 192 && ip[1] == 168 || ip[0] == 169 && ip[1] == 254)
}
func lower16BitPrivateIP(interfaceAddrs types.InterfaceAddrs) (int, error) {
ip, err := privateIPv4(interfaceAddrs)
if err != nil {
return 0, err
}
return int(ip[2])<<8 + int(ip[3]), nil
}
func (sf *Sonyflake) ToTime(id int64) time.Time {
return time.Unix(0, (sf.startTime+Time(id))*sf.timeUnit)
}
// Time returns the Sonyflake time when the given ID was generated.
func Time(id int64) int64 {
return id >> (BitLenSequence + BitLenMachine)
}
// SequenceNumber returns the sequence number of a Sonyflake ID.
func SequenceNumber(id int64) int {
const maskSequence = int64((1<<BitLenSequence - 1) << BitLenMachine)
return int((id & maskSequence) >> BitLenMachine)
}
// MachineID returns the machine ID of a Sonyflake ID.
func MachineID(id int64) int {
const maskMachine = int64(1<<BitLenMachine - 1)
return int(id & maskMachine)
}
// Decompose returns a set of Sonyflake ID parts.
func Decompose(id int64) map[string]int64 {
time := Time(id)
sequence := SequenceNumber(id)
machine := MachineID(id)
return map[string]int64{
"id": id,
"time": time,
"sequence": int64(sequence),
"machine": int64(machine),
}
}
+331
View File
@@ -0,0 +1,331 @@
package sonyflake
import (
"errors"
"fmt"
"net"
"runtime"
"testing"
"time"
"github.com/sony/sonyflake/v2/mock"
"github.com/sony/sonyflake/v2/types"
)
func TestNew(t *testing.T) {
errGetMachineID := fmt.Errorf("failed to get machine id")
testCases := []struct {
name string
settings Settings
err error
}{
{
name: "invalid time unit",
settings: Settings{
TimeUnit: time.Microsecond,
},
err: ErrInvalidTimeUnit,
},
{
name: "start time ahead",
settings: Settings{
StartTime: time.Now().Add(time.Minute),
},
err: ErrStartTimeAhead,
},
{
name: "cannot get machine id",
settings: Settings{
MachineID: func() (int, error) {
return 0, errGetMachineID
},
},
err: errGetMachineID,
},
{
name: "invalid machine id",
settings: Settings{
CheckMachineID: func(int) bool {
return false
},
},
err: ErrInvalidMachineID,
},
{
name: "success",
settings: Settings{},
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
sf, err := New(tc.settings)
if !errors.Is(err, tc.err) {
t.Fatalf("unexpected error: %v", err)
}
if err == nil && sf == nil {
t.Fatal("sonyflake instance must be created")
}
})
}
}
func newSonyflake(t *testing.T, st Settings) *Sonyflake {
sf, err := New(st)
if err != nil {
t.Fatalf("failed to create sonyflake: %v", err)
}
return sf
}
func nextID(t *testing.T, sf *Sonyflake) int64 {
id, err := sf.NextID()
if err != nil {
t.Fatalf("failed to generate id: %v", err)
}
return id
}
func defaultMachineID(t *testing.T) int {
ip, err := lower16BitPrivateIP(defaultInterfaceAddrs)
if err != nil {
t.Fatalf("failed to get private ip address: %v", err)
}
return ip
}
func TestNextID(t *testing.T) {
sf := newSonyflake(t, Settings{StartTime: time.Now()})
sleepTime := int64(50)
time.Sleep(time.Duration(sleepTime * sf.timeUnit))
id := nextID(t, sf)
actualTime := Time(id)
if actualTime < sleepTime || actualTime > sleepTime+1 {
t.Errorf("unexpected time: %d", actualTime)
}
actualSequence := SequenceNumber(id)
if actualSequence != 0 {
t.Errorf("unexpected sequence: %d", actualSequence)
}
actualMachine := MachineID(id)
if actualMachine != defaultMachineID(t) {
t.Errorf("unexpected machine: %d", actualMachine)
}
fmt.Println("sonyflake id:", id)
fmt.Println("decompose:", Decompose(id))
}
func TestNextID_InSequence(t *testing.T) {
now := time.Now()
sf := newSonyflake(t, Settings{
TimeUnit: time.Millisecond,
StartTime: now,
})
startTime := sf.toInternalTime(now)
machineID := int64(defaultMachineID(t))
var numID int
var lastID int64
var maxSeq int64
currentTime := startTime
for currentTime-startTime < 100 {
id := nextID(t, sf)
currentTime = sf.toInternalTime(time.Now())
numID++
if id == lastID {
t.Fatal("duplicated id")
}
if id < lastID {
t.Fatal("must increase with time")
}
lastID = id
parts := Decompose(id)
actualTime := parts["time"]
overtime := startTime + actualTime - currentTime
if overtime > 0 {
t.Errorf("unexpected overtime: %d", overtime)
}
actualSequence := parts["sequence"]
if actualSequence > maxSeq {
maxSeq = actualSequence
}
actualMachine := parts["machine"]
if actualMachine != machineID {
t.Errorf("unexpected machine: %d", actualMachine)
}
}
if maxSeq != 1<<BitLenSequence-1 {
t.Errorf("unexpected max sequence: %d", maxSeq)
}
fmt.Println("max sequence:", maxSeq)
fmt.Println("number of id:", numID)
}
func TestNextID_InParallel(t *testing.T) {
sf1 := newSonyflake(t, Settings{MachineID: func() (int, error) { return 1, nil }})
sf2 := newSonyflake(t, Settings{MachineID: func() (int, error) { return 2, nil }})
numCPU := runtime.NumCPU()
runtime.GOMAXPROCS(numCPU)
fmt.Println("number of cpu:", numCPU)
consumer := make(chan int64)
const numID = 1000
generate := func(sf *Sonyflake) {
for i := 0; i < numID; i++ {
id := nextID(t, sf)
consumer <- id
}
}
var numGenerator int
for i := 0; i < numCPU/2; i++ {
go generate(sf1)
go generate(sf2)
numGenerator += 2
}
set := make(map[int64]struct{})
for i := 0; i < numID*numGenerator; i++ {
id := <-consumer
if _, ok := set[id]; ok {
t.Fatal("duplicated id")
}
set[id] = struct{}{}
}
fmt.Println("number of id:", len(set))
}
func pseudoSleep(sf *Sonyflake, period time.Duration) {
sf.startTime -= int64(period) / sf.timeUnit
}
func TestNextID_ReturnsError(t *testing.T) {
sf := newSonyflake(t, Settings{StartTime: time.Now()})
year := time.Duration(365*24) * time.Hour
pseudoSleep(sf, time.Duration(174)*year)
nextID(t, sf)
pseudoSleep(sf, time.Duration(1)*year)
_, err := sf.NextID()
if err == nil {
t.Errorf("time is not over")
}
}
func TestToTime(t *testing.T) {
start := time.Now()
sf := newSonyflake(t, Settings{
TimeUnit: time.Millisecond,
StartTime: start,
})
id := nextID(t, sf)
tm := sf.ToTime(id)
diff := tm.Sub(start)
if diff < 0 || diff >= time.Duration(sf.timeUnit) {
t.Errorf("unexpected time: %v", tm)
}
}
func TestPrivateIPv4(t *testing.T) {
testCases := []struct {
description string
expected net.IP
interfaceAddrs types.InterfaceAddrs
error error
}{
{
description: "returns an error",
expected: nil,
interfaceAddrs: mock.NewFailingInterfaceAddrs(),
error: mock.ErrFailedToGetAddresses,
},
{
description: "empty address list",
expected: nil,
interfaceAddrs: mock.NewNilInterfaceAddrs(),
error: ErrNoPrivateAddress,
},
{
description: "success",
expected: net.IP{192, 168, 0, 1},
interfaceAddrs: mock.NewSuccessfulInterfaceAddrs(),
error: nil,
},
}
for _, tc := range testCases {
t.Run(tc.description, func(t *testing.T) {
actual, err := privateIPv4(tc.interfaceAddrs)
if !errors.Is(err, tc.error) {
t.Fatalf("unexpected error: %v", err)
}
if !net.IP.Equal(actual, tc.expected) {
t.Errorf("unexpected ip: %s", actual)
}
})
}
}
func TestLower16BitPrivateIP(t *testing.T) {
testCases := []struct {
description string
expected int
interfaceAddrs types.InterfaceAddrs
error error
}{
{
description: "returns an error",
expected: 0,
interfaceAddrs: mock.NewFailingInterfaceAddrs(),
error: mock.ErrFailedToGetAddresses,
},
{
description: "empty address list",
expected: 0,
interfaceAddrs: mock.NewNilInterfaceAddrs(),
error: ErrNoPrivateAddress,
},
{
description: "success",
expected: 1,
interfaceAddrs: mock.NewSuccessfulInterfaceAddrs(),
error: nil,
},
}
for _, tc := range testCases {
t.Run(tc.description, func(t *testing.T) {
actual, err := lower16BitPrivateIP(tc.interfaceAddrs)
if !errors.Is(err, tc.error) {
t.Fatalf("unexpected error: %v", err)
}
if actual != tc.expected {
t.Errorf("unexpected ip: %d", actual)
}
})
}
}
+8
View File
@@ -0,0 +1,8 @@
// Package types defines type signatures used throughout sonyflake.
// This provides the ability to mock out imports.
package types
import "net"
// InterfaceAddrs defines the interface used for retrieving network addresses.
type InterfaceAddrs func() ([]net.Addr, error)